| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "os" |
| 10 | "strings" |
| 11 | "sync" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/activator" |
| 15 | "github.com/carlosframework/platform/internal/lease" |
| 16 | "github.com/carlosframework/platform/internal/moves" |
| 17 | "github.com/carlosframework/platform/internal/provision" |
| 18 | "github.com/carlosframework/platform/internal/release" |
| 19 | ) |
| 20 | |
| 21 | // The pre-copy of an offline move (offline-staging): while the instance |
| 22 | // keeps serving, one source box (the stager) runs mirror rounds of its |
| 23 | // replica into the target's staging prefix, so the seal's own round, with |
| 24 | // the instance offline, copies only what changed since. |
| 25 | // |
| 26 | // A stage round is the seal's round with nothing else: the same source |
| 27 | // (instances/<host>/ in this box's bucket), the same destination |
| 28 | // (moves/<host>/ in the target's data bucket), the same receipts |
| 29 | // (move-receipts/<host>.json in this box's bucket) and the same mirror pair. |
| 30 | // The staging prefix and the receipts are keyed by host, not by move, so |
| 31 | // the seal finds what the stage copied with no hand-over: lease.MirrorPrefix |
| 32 | // skips an object only when its ETag equals the source's or a receipt for |
| 33 | // this exact pair still matches both sides, deletes every staged key the |
| 34 | // source no longer has (litestream's retention between stage and seal), |
| 35 | // and copies everything else. Nothing a stage round left behind is trusted |
| 36 | // beyond that, so a stale prefix is corrected, never sealed as it is. |
| 37 | // |
| 38 | // The round holds nothing, fences nothing and takes no lease: it reads the |
| 39 | // replica (its listing and the copies' reads) and writes only the target's |
| 40 | // staging prefix and this box's receipts object. The lease, lock.json |
| 41 | // included, is never touched (MirrorPrefix skips lock.json on both sides). |
| 42 | // |
| 43 | // One round runs per stage request: the record's staged_at, which the stage |
| 44 | // door sets and every later stage moves forward (provision.OfflineRestage). |
| 45 | // The stager report names the staged_at its round answered |
| 46 | // (StageRequest), and a pass whose stager report already answers the |
| 47 | // record's runs nothing, so an idle staging costs one report read a pass and |
| 48 | // no S3 request; a failed round is retried by running stage again. Rounds |
| 49 | // stop when the record leaves staging: the pass that runs a round holds the |
| 50 | // worker's pass lock, so this box's stopped report, which the seal door |
| 51 | // waits for, is written only after the last round it ran has ended. |
| 52 | |
| 53 | // capOfflineStage1 is the edges-file capability a box publishes beside |
| 54 | // offline-move-1 when it can stage: run the pre-copy as the stager, and |
| 55 | // clean it up on abort. The console's stage door refuses a region holding a |
| 56 | // box that does not publish it. |
| 57 | const capOfflineStage1 = "offline-stage-1" |
| 58 | |
| 59 | // offlineStageStep is the stager's step in offline-staging: one round, then |
| 60 | // the stager report. The stager is the box the stage door pinned on the |
| 61 | // record (MoveState.Stager); no other box ever writes this move's staging |
| 62 | // prefix, whatever the fleet looks like now (Astra round 1, finding 2). |
| 63 | func (w *movesWorker) offlineStageStep(ctx context.Context, s release.Store, rec provision.Record) error { |
| 64 | cfg := w.cfg |
| 65 | m, host := rec.Move, rec.Host |
| 66 | if cfg.Box == "" { |
| 67 | return errors.New("offline move: this box has no label to report under") |
| 68 | } |
| 69 | if m.Stager == "" || m.Stager != cfg.Box { |
| 70 | return nil |
| 71 | } |
| 72 | // The record again, now, under this box's pass lock: the pass's listing |
| 73 | // may predate the start, and once the move has left staging no round |
| 74 | // ever starts (Astra round 2, finding 1). |
| 75 | cur, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| 76 | if err != nil { |
| 77 | return fmt.Errorf("offline move %s: read the record before a stage round: %w", m.ID, err) |
| 78 | } |
| 79 | if cur.Move == nil || cur.Move.ID != m.ID || cur.Move.State != provision.OfflineStaging { |
| 80 | return nil |
| 81 | } |
| 82 | rec, m = *cur, cur.Move |
| 83 | var request time.Time |
| 84 | if m.StagedAt != nil { |
| 85 | request = m.StagedAt.UTC() |
| 86 | } |
| 87 | prev, err := w.ownStagerReport(ctx, s, rec) |
| 88 | if err != nil { |
| 89 | return fmt.Errorf("offline move %s: read the stager report: %w", m.ID, err) |
| 90 | } |
| 91 | if prev != nil && prev.StageRequest.Equal(request) && (prev.Phase == moves.OfflinePhaseStaged || prev.Phase == moves.OfflinePhaseFailed) { |
| 92 | return nil // this request has had its round |
| 93 | } |
| 94 | r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleStager, Phase: moves.OfflinePhaseStaged, StageRequest: request} |
| 95 | p, err := w.offlineStageRound(ctx, s, rec) |
| 96 | r.At = cfg.now().UTC() |
| 97 | if err != nil { |
| 98 | // The progress stays what the last good round found; the error is a |
| 99 | // box diagnostic the status view shows as it is. |
| 100 | if prev != nil { |
| 101 | p = stageProgress{replicaObjects: prev.ReplicaObjects, replicaBytes: prev.ReplicaBytes, stagedObjects: prev.StagedObjects, stagedBytes: prev.StagedBytes} |
| 102 | } |
| 103 | p.fill(&r) |
| 104 | r.Phase, r.StageError = moves.OfflinePhaseFailed, err.Error() |
| 105 | return errors.Join(fmt.Errorf("offline move %s: staging %s: %w", m.ID, host, err), writeOfflineReport(ctx, s, r)) |
| 106 | } |
| 107 | p.fill(&r) |
| 108 | if p.roundObjects > 0 { |
| 109 | log.Printf("moves: %s: offline move %s: staged round: %d object(s) changed, %d bytes copied; %d of %d objects staged", host, m.ID, p.roundObjects, p.roundBytes, p.stagedObjects, p.replicaObjects) |
| 110 | } |
| 111 | return writeOfflineReport(ctx, s, r) |
| 112 | } |
| 113 | |
| 114 | // stageProgress is what one stage round found. |
| 115 | type stageProgress struct { |
| 116 | replicaObjects, stagedObjects, roundObjects int |
| 117 | replicaBytes, stagedBytes, roundBytes int64 |
| 118 | } |
| 119 | |
| 120 | func (p stageProgress) fill(r *moves.OfflineReport) { |
| 121 | r.ReplicaObjects, r.ReplicaBytes = p.replicaObjects, p.replicaBytes |
| 122 | r.StagedObjects, r.StagedBytes = p.stagedObjects, p.stagedBytes |
| 123 | r.RoundObjects, r.RoundBytes = p.roundObjects, p.roundBytes |
| 124 | } |
| 125 | |
| 126 | // offlineStageRound is one pre-copy round and the listing that measures it. |
| 127 | // It writes only to the staging destination the stage fixed on the record |
| 128 | // (Astra round 2, finding 2), and only while the target still publishes |
| 129 | // exactly that bucket and region; a target that publishes anything else is |
| 130 | // refused, never followed. The round's delete phase reads the record again |
| 131 | // first and stops if the move has left staging (finding 1). |
| 132 | func (w *movesWorker) offlineStageRound(ctx context.Context, s release.Store, rec provision.Record) (stageProgress, error) { |
| 133 | cfg := w.cfg |
| 134 | host, m := rec.Host, rec.Move |
| 135 | if m.StageBucket == "" { |
| 136 | return stageProgress{}, errors.New("the record names no staging bucket") |
| 137 | } |
| 138 | tgt, err := w.offlineTarget(ctx, s, m) |
| 139 | if err != nil { |
| 140 | return stageProgress{}, err |
| 141 | } |
| 142 | if tgt.DataBucket != m.StageBucket || tgt.DataBucketRegion != m.StageRegion { |
| 143 | return stageProgress{}, fmt.Errorf("the target now publishes %s (%s), not the staging bucket %s (%s) the stage fixed", tgt.DataBucket, tgt.DataBucketRegion, m.StageBucket, m.StageRegion) |
| 144 | } |
| 145 | bucket, loc, _, err := cfg.tombstoneAt(host) |
| 146 | if err != nil { |
| 147 | return stageProgress{}, err |
| 148 | } |
| 149 | // The activator promotes a host's moves/<host>/ in its OWN bucket over |
| 150 | // its replica on every restore (promoteStagedReplica: staging wins). A |
| 151 | // staging bucket that is this box's own would have the serving |
| 152 | // instance restore from a copy behind its replica, so the round refuses |
| 153 | // (the stage door refuses it first). |
| 154 | if m.StageBucket == bucket { |
| 155 | return stageProgress{}, fmt.Errorf("the staging bucket is this box's own bucket %s; a staged copy there would replace the serving replica on its next wake", bucket) |
| 156 | } |
| 157 | mirror, err := w.offlineMirror(ctx, m.StageRegion) |
| 158 | if err != nil { |
| 159 | return stageProgress{}, err |
| 160 | } |
| 161 | gated := &stageDeleteGate{MirrorStore: mirror, check: func(ctx context.Context) error { |
| 162 | cur, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| 163 | if err != nil { |
| 164 | return fmt.Errorf("read the record before the delete phase: %w", err) |
| 165 | } |
| 166 | if cur.Move == nil || cur.Move.ID != m.ID || cur.Move.State != provision.OfflineStaging { |
| 167 | return errors.New("the move left staging during the round; its deletes are not made") |
| 168 | } |
| 169 | return nil |
| 170 | }} |
| 171 | srcPrefix := loc + "/" |
| 172 | dstPrefix := activator.MovesPrefix(host) |
| 173 | receipts := lease.Receipts{Bucket: bucket, Key: activator.MoveReceiptsKey(host)} |
| 174 | round, err := lease.MirrorPrefix(ctx, gated, bucket, srcPrefix, m.StageBucket, dstPrefix, receipts) |
| 175 | if err != nil { |
| 176 | return stageProgress{}, err |
| 177 | } |
| 178 | source, err := mirror.List(ctx, bucket, srcPrefix) |
| 179 | if err != nil { |
| 180 | return stageProgress{}, fmt.Errorf("list the replica: %w", err) |
| 181 | } |
| 182 | staged, err := mirror.List(ctx, m.StageBucket, dstPrefix) |
| 183 | if err != nil { |
| 184 | return stageProgress{}, fmt.Errorf("list the staged copy: %w", err) |
| 185 | } |
| 186 | p := stageProgress{roundObjects: round.Changed, roundBytes: round.BytesCopied} |
| 187 | for k, o := range source { |
| 188 | rel := strings.TrimPrefix(k, srcPrefix) |
| 189 | if rel == "lock.json" { |
| 190 | continue |
| 191 | } |
| 192 | p.replicaObjects++ |
| 193 | p.replicaBytes += o.Size |
| 194 | if d, ok := staged[dstPrefix+rel]; ok && d.Size == o.Size { |
| 195 | p.stagedObjects++ |
| 196 | p.stagedBytes += o.Size |
| 197 | } |
| 198 | } |
| 199 | return p, nil |
| 200 | } |
| 201 | |
| 202 | // stageDeleteGate is a stage round's mirror with one check before the |
| 203 | // delete phase: lease.MirrorPrefix deletes only after every copy has |
| 204 | // landed, so the first Delete is that phase's start, and check runs there |
| 205 | // once for all of the round's deletes. Its error fails every delete, so |
| 206 | // the round deletes nothing. |
| 207 | type stageDeleteGate struct { |
| 208 | lease.MirrorStore |
| 209 | check func(ctx context.Context) error |
| 210 | once sync.Once |
| 211 | err error |
| 212 | } |
| 213 | |
| 214 | func (g *stageDeleteGate) Delete(ctx context.Context, bucket, key string) error { |
| 215 | g.once.Do(func() { g.err = g.check(ctx) }) |
| 216 | if g.err != nil { |
| 217 | return g.err |
| 218 | } |
| 219 | return g.MirrorStore.Delete(ctx, bucket, key) |
| 220 | } |
| 221 | |
| 222 | // offlineStagerStopped is the pinned stager's stop acknowledgement, written |
| 223 | // in stopping and every later state up to landed, on every pass and every |
| 224 | // reassert tick, always under this box's pass lock (Astra round 2, finding |
| 225 | // 1). The record it was handed has left staging, which never comes back, |
| 226 | // so no stage round starts from here on (offlineStageStep reads the record |
| 227 | // again before one), and the round that holds the pass lock has ended |
| 228 | // before this runs. The seal door waits for it, fresh. The progress of the |
| 229 | // last round is kept. |
| 230 | func (w *movesWorker) offlineStagerStopped(ctx context.Context, s release.Store, rec provision.Record) error { |
| 231 | m := rec.Move |
| 232 | if m.StagedAt == nil || m.Stager == "" || m.Stager != w.cfg.Box { |
| 233 | return nil |
| 234 | } |
| 235 | prev, err := w.ownStagerReport(ctx, s, rec) |
| 236 | if err != nil { |
| 237 | return fmt.Errorf("offline move %s: read the stager report: %w", m.ID, err) |
| 238 | } |
| 239 | r := moves.OfflineReport{Op: m.ID, Host: rec.Host, Box: w.cfg.Box, Role: moves.OfflineRoleStager, Phase: moves.OfflinePhaseStopped, |
| 240 | At: w.cfg.now().UTC(), StageRequest: m.StagedAt.UTC()} |
| 241 | if prev != nil { |
| 242 | stageProgress{replicaObjects: prev.ReplicaObjects, replicaBytes: prev.ReplicaBytes, stagedObjects: prev.StagedObjects, |
| 243 | stagedBytes: prev.StagedBytes, roundObjects: prev.RoundObjects, roundBytes: prev.RoundBytes}.fill(&r) |
| 244 | } |
| 245 | return writeOfflineReport(ctx, s, r) |
| 246 | } |
| 247 | |
| 248 | // ownStagerReport is this box's stager report of the move, nil when it has |
| 249 | // written none. One that does not read as a report is an error. |
| 250 | func (w *movesWorker) ownStagerReport(ctx context.Context, s release.Store, rec provision.Record) (*moves.OfflineReport, error) { |
| 251 | key := moves.OfflineReportKey(rec.Host, rec.Move.ID, moves.OfflineRoleStager, w.cfg.Box) |
| 252 | if key == "" { |
| 253 | return nil, fmt.Errorf("offline move %s: %s has no stager report key", rec.Move.ID, rec.Host) |
| 254 | } |
| 255 | raw, _, err := s.Get(ctx, key) |
| 256 | if errors.Is(err, os.ErrNotExist) { |
| 257 | return nil, nil |
| 258 | } |
| 259 | if err != nil { |
| 260 | return nil, err |
| 261 | } |
| 262 | var r moves.OfflineReport |
| 263 | if err := json.Unmarshal(raw, &r); err != nil { |
| 264 | return nil, fmt.Errorf("%s: %w", key, err) |
| 265 | } |
| 266 | return &r, nil |
| 267 | } |
| 268 | |
| 269 | // offlineUnstage is the stager's part of an abort of a staged move. A move |
| 270 | // that carries staged_at may have staged copies whatever its reports say: a |
| 271 | // round can land copies and die before its report (Astra round 1, finding |
| 272 | // 1). So the pinned stager, and only it, deletes the target's staging |
| 273 | // prefix and the copy receipts and says so in its stager report (phase |
| 274 | // reverted), once, before its source step reports reverted. Its own last |
| 275 | // round has ended by then: rounds run only in staging, and both run under |
| 276 | // this box's pass lock. The console's finish of a staged move waits for |
| 277 | // that report (finding 2). A move that was never staged does nothing here, |
| 278 | // so an abort from stopping is what it was. The prefix deleted is this |
| 279 | // host's in the target's bucket, never this box's own bucket, where a |
| 280 | // moves/<host>/ prefix is not this move's (the stage round refuses to write |
| 281 | // there for the same reason). |
| 282 | func (w *movesWorker) offlineUnstage(ctx context.Context, s release.Store, rec provision.Record) error { |
| 283 | cfg := w.cfg |
| 284 | m, host := rec.Move, rec.Host |
| 285 | if m.StagedAt == nil || m.Stager == "" || m.Stager != cfg.Box { |
| 286 | return nil |
| 287 | } |
| 288 | own, err := w.ownStagerReport(ctx, s, rec) |
| 289 | if err != nil { |
| 290 | return fmt.Errorf("offline move %s: read the stager report: %w", m.ID, err) |
| 291 | } |
| 292 | if own != nil && own.Phase == moves.OfflinePhaseReverted { |
| 293 | return nil |
| 294 | } |
| 295 | // The destination the stage fixed on the record, never what the |
| 296 | // target publishes now (Astra round 2, finding 2): reverted is written |
| 297 | // only after that prefix is deleted. A stored bucket that is this box's |
| 298 | // own was refused at the stage, so it is never deleted from here, and |
| 299 | // nothing is acknowledged. |
| 300 | if m.StageBucket == "" { |
| 301 | return fmt.Errorf("offline move %s: the record names no staging bucket to clean", m.ID) |
| 302 | } |
| 303 | if m.StageBucket == cfg.OwnBucket { |
| 304 | return fmt.Errorf("offline move %s: the stored staging bucket %s is this box's own; nothing is deleted or acknowledged", m.ID, m.StageBucket) |
| 305 | } |
| 306 | mirror, err := w.offlineMirror(ctx, m.StageRegion) |
| 307 | if err != nil { |
| 308 | return err |
| 309 | } |
| 310 | prefix := activator.MovesPrefix(host) |
| 311 | if prefix == "moves/" || !strings.HasPrefix(prefix, "moves/") { |
| 312 | return fmt.Errorf("offline move %s: no staging prefix for %s", m.ID, host) |
| 313 | } |
| 314 | if err := lease.DeletePrefix(ctx, mirror, m.StageBucket, prefix); err != nil { |
| 315 | return fmt.Errorf("offline move %s: delete %s/%s: %w", m.ID, m.StageBucket, prefix, err) |
| 316 | } |
| 317 | log.Printf("moves: %s: offline move %s: the staged copy in %s/%s is deleted", host, m.ID, m.StageBucket, prefix) |
| 318 | clearOfflineReceipts(ctx, mirror, cfg.OwnBucket, host) |
| 319 | r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleStager, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC()} |
| 320 | return writeOfflineReport(ctx, s, r) |
| 321 | } |
| 322 | |
amadan