amadan
moves offline: stage, a pre-copy of the replica while the instance serves
An offline move can now begin in offline-staging (provision.OfflineStage, the console's POST .../move/offline/stage). Nothing holds, fences or tombstones the instance there. The source region's stager (the lexically first box that publishes offline-stage-1, moves.OfflineStager) runs a mirror round of instances/<host>/ into the target's moves/<host>/ on every moves pass, and writes a stager report with its progress. A second stage writes nothing and bumps the serial for another round now. The stage round is the seal's round with nothing else around it: the same source, destination, receipts object and mirror pair, all keyed by host, so the seal's own round finds what the stage copied and skips it by receipt or ETag, deletes whatever the source dropped in between, and copies the rest. A stale staged prefix is corrected the same way, never sealed as it stands. The stage round refuses a target that publishes the source's own bucket, where the activator's staging-wins promotion would replace the serving replica. start on a staged record is OfflineStartStaged (staging to stopping, requested_at moved to the start so offline time counts from there, and always after the new staged_at). abort from staging is the ordinary revert; the stager deletes the staged copy and the receipts before it reports reverted, so the finish waits for it, and a move aborted while staging records no offline time. The status view carries the stage's progress (stage). Box publishes offline-stage-1 beside offline-move-1. Copy rows follow in the CLI commit. Census: TestClaimConsoleRouteCensus fails on the new route (the census inventory is left for the ruling).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
16 files changed,
+1279
−42
internal/console/apiinstancemove_offline.go+166 −21internal/console/apiinstancemove_offline_stage_test.go+183 −0internal/console/apiinstancemove_offline_test.go+6 −1internal/console/authcli.go+1 −0internal/console/claim_invariant_fixtures_test.go+7 −0internal/console/svcroutes_test.go+1 −0internal/host/edgesfile.go+3 −2internal/host/movesoffline.go+12 −0internal/host/movesoffline_stage.go+251 −0internal/host/movesoffline_stage_test.go+322 −0internal/host/movesoffline_test.go+4 −4internal/moves/offline.go+45 −5internal/moves/offline_test.go+30 −0internal/provision/offline_move.go+63 −9internal/provision/offline_stage_test.go+178 −0internal/provision/provision.go+7 −0
diff --git a/internal/console/apiinstancemove_offline.go b/internal/console/apiinstancemove_offline.go| index 74d178a72..8536b8151 100644 |
| --- a/internal/console/apiinstancemove_offline.go |
| +++ b/internal/console/apiinstancemove_offline.go |
| @@ -29,7 +29,8 @@ import ( |
| // offline-operator-move.md, Task 6). One hibernating instance moves between |
| // two regions with planned downtime, an operator at each step: |
| // |
| -// POST .../instances/{host}/move/offline {"region":"uk"} start |
| +// POST .../instances/{host}/move/offline/stage {"region":"uk"} pre-copy: -> staging |
| +// POST .../instances/{host}/move/offline {"region":"uk"} start (staging -> stopping) |
| // POST .../instances/{host}/move/offline/seal stopping -> sealing |
| // POST .../instances/{host}/move/offline/flip sealed -> landing |
| // POST .../instances/{host}/move/offline/open landed -> open |
| @@ -52,6 +53,12 @@ import ( |
| // enforced guard are all off (internal/host, Task 3). |
| const offlineMoveCapability = "offline-move-1" |
| +// offlineStageCapability is the one a box publishes beside it when it can |
| +// stage (internal/host movesoffline_stage.go): run the pre-copy as the |
| +// stager and remove it on abort. A box without it treats a staging record |
| +// as nothing to do, so the stage door refuses a region holding one. |
| +const offlineStageCapability = "offline-stage-1" |
| + |
| // offlineReportFresh is how recently a box's report must be written for a |
| // door to believe it. Boxes rewrite their reports on every moves pass, and a |
| // source or sealer box every 15 s or so while its report is one a door reads |
| @@ -469,62 +476,64 @@ func (s *Server) offlineClaimEvidence(ctx context.Context, d offlineDoor) *offli |
| return nil |
| } |
| -func (s *Server) handleAPIInstanceOfflineStart(w http.ResponseWriter, r *http.Request) { |
| - d, ok := s.offlineGate(w, r) |
| - if !ok { |
| - return |
| - } |
| +// offlineBeginChecks is what start and stage check before either writes: |
| +// the instance's class, its claims, the region asked for, the app's |
| +// regions and the fleet's box rules (offlineSelect). It answers the region |
| +// code and the fleet it read, or false once it has answered the request. |
| +// A staged move is checked again in full when it starts: the fleet and the |
| +// claims may have changed since the stage. |
| +func (s *Server) offlineBeginChecks(w http.ResponseWriter, r *http.Request, d offlineDoor) (string, offlineFleet, bool) { |
| rec, host := d.rec, d.host |
| if s.refuseSeedMove(w, rec) { |
| - return |
| + return "", offlineFleet{}, false |
| } |
| if rec.Role != "" { |
| jsonError(w, http.StatusBadRequest, "a replica does not move. delete it and create one in the new region") |
| - return |
| + return "", offlineFleet{}, false |
| } |
| if len(rec.Replicas) > 0 { |
| jsonError(w, http.StatusBadRequest, "a home with replicas does not move. remove its replicas first") |
| - return |
| + return "", offlineFleet{}, false |
| } |
| if rec.Steering != "" || rec.Unit != "" { |
| jsonError(w, http.StatusBadRequest, fmt.Sprintf("%s is steered or runs under a unit. offline moves cannot carry either yet", host)) |
| - return |
| + return "", offlineFleet{}, false |
| } |
| ctx := r.Context() |
| if ref := s.offlineClaimEvidence(ctx, d); ref != nil { |
| writeRefusal(w, ref) |
| - return |
| + return "", offlineFleet{}, false |
| } |
| var req struct { |
| Region string `json:"region"` |
| } |
| if !readJSON(w, r, &req) { |
| - return |
| + return "", offlineFleet{}, false |
| } |
| region := strings.TrimSpace(req.Region) |
| to := s.homeRegionIn(region) |
| if region == "" || !provision.ValidRegion(to) { |
| jsonError(w, http.StatusBadRequest, "invalid region") |
| - return |
| + return "", offlineFleet{}, false |
| } |
| if to == rec.Region { |
| jsonError(w, http.StatusConflict, "the instance is already in "+region) |
| - return |
| + return "", offlineFleet{}, false |
| } |
| m, err := provision.GetManifest(ctx, s.cfg.Releases, d.acct.Sqid, d.app) |
| if err != nil && !errors.Is(err, os.ErrNotExist) { |
| log.Printf("console: offline move %s: manifest: %v", host, err) |
| jsonError(w, http.StatusServiceUnavailable, "could not read the app's instances manifest. try again") |
| - return |
| + return "", offlineFleet{}, false |
| } |
| if m != nil && !s.moveRegionAllowed(m.Regions, to) { |
| jsonError(w, http.StatusBadRequest, "region "+region+" is not one of this app's regions: "+strings.Join(m.Regions, ", ")) |
| - return |
| + return "", offlineFleet{}, false |
| } |
| fleet, ok := s.readOfflineFleet(ctx) |
| if !ok { |
| jsonError(w, http.StatusConflict, "the fleet's edges files could not all be read. try again") |
| - return |
| + return "", offlineFleet{}, false |
| } |
| if to == "" || rec.Region == "" { |
| // A box that sets CARLOS_REGION to the home's name is home to this |
| @@ -533,22 +542,105 @@ func (s *Server) handleAPIInstanceOfflineStart(w http.ResponseWriter, r *http.Re |
| for _, f := range fleet.snap.Files { |
| if f.Placement != "" && homePlacementCode(s.moveHomeName(), f.Placement) == "" { |
| jsonError(w, http.StatusConflict, f.Box+" sets CARLOS_REGION, so it cannot take a move into the home region") |
| - return |
| + return "", offlineFleet{}, false |
| } |
| } |
| } |
| if ref := s.offlineSelect(ctx, fleet, host, rec.Region, to); ref != nil { |
| writeRefusal(w, ref) |
| + return "", offlineFleet{}, false |
| + } |
| + return to, fleet, true |
| +} |
| + |
| +// offlineStagedFor is the refusal of a start or a stage that names another |
| +// region than the one the instance is staged for (MV-O179). |
| +func (s *Server) offlineStagedFor(host, to string) *offlineRefusal { |
| + return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s is staged for %s. abort the move to move it to another region", host, s.offlineName(to))} |
| +} |
| + |
| +// offlineStaged is the record's offline move when it is staging. |
| +func offlineStaged(rec *provision.Record) (*provision.MoveState, bool) { |
| + if mv := rec.Move; mv.Offline() && mv.State == provision.OfflineStaging { |
| + return mv, true |
| + } |
| + return nil, false |
| +} |
| + |
| +func (s *Server) handleAPIInstanceOfflineStart(w http.ResponseWriter, r *http.Request) { |
| + d, ok := s.offlineGate(w, r) |
| + if !ok { |
| return |
| } |
| + to, _, ok := s.offlineBeginChecks(w, r, d) |
| + if !ok { |
| + return |
| + } |
| + // A staged instance starts the move it is staged for: staging to |
| + // stopping, the instance offline from now. The lock is the stage's. |
| + if mv, staged := offlineStaged(d.rec); staged { |
| + if to != mv.To { |
| + writeRefusal(w, s.offlineStagedFor(d.host, mv.To)) |
| + return |
| + } |
| + s.offlineAdvance(w, r, d, provision.OfflineStartStaged(mv.ID, d.addr, time.Now())) |
| + return |
| + } |
| + s.offlineBegin(w, r, d, func(id string) func(*provision.Record) error { |
| + return provision.OfflineStart(id, d.rec.Region, to, d.addr, time.Now()) |
| + }) |
| +} |
| + |
| +// handleAPIInstanceOfflineStage is the pre-copy door. On a record at rest it |
| +// creates the move in offline-staging, checked exactly as start is and |
| +// under the same one-move lock, and the source region's stager begins |
| +// copying on the serial bump. On a record already staging for that region |
| +// it writes nothing and bumps the serial, so the stager runs another round |
| +// now; each round copies only what is new (the receipts). |
| +func (s *Server) handleAPIInstanceOfflineStage(w http.ResponseWriter, r *http.Request) { |
| + d, ok := s.offlineGate(w, r) |
| + if !ok { |
| + return |
| + } |
| + to, fleet, ok := s.offlineBeginChecks(w, r, d) |
| + if !ok { |
| + return |
| + } |
| + ctx := r.Context() |
| + if mv, staged := offlineStaged(d.rec); staged { |
| + if to != mv.To { |
| + writeRefusal(w, s.offlineStagedFor(d.host, mv.To)) |
| + return |
| + } |
| + s.bumpSerial(ctx) |
| + writeJSON(w, http.StatusAccepted, s.offlineViewOf(ctx, d.rec)) |
| + return |
| + } |
| + for _, b := range append(fleet.boxesAt(d.rec.Region), fleet.boxesAt(to)...) { |
| + if !slices.Contains(b.Capabilities, offlineStageCapability) { |
| + jsonError(w, http.StatusConflict, fmt.Sprintf("%s cannot stage an offline move. it runs an older release", b.Box)) |
| + return |
| + } |
| + } |
| + s.offlineBegin(w, r, d, func(id string) func(*provision.Record) error { |
| + return provision.OfflineStage(id, d.rec.Region, to, d.addr, time.Now()) |
| + }) |
| +} |
| + |
| +// offlineBegin mints the move's id and writes the transition begin makes |
| +// with it (OfflineStart or OfflineStage) under the one-move lock, then |
| +// bumps the serial and answers 202 with the status view. |
| +func (s *Server) offlineBegin(w http.ResponseWriter, r *http.Request, d offlineDoor, begin func(id string) func(*provision.Record) error) { |
| + ctx := r.Context() |
| + host := d.host |
| id, err := moves.NewMoveID(rand.Reader) |
| if err != nil { |
| log.Printf("console: offline move %s: move id: %v", host, err) |
| jsonError(w, http.StatusBadGateway, "could not mint a move id") |
| return |
| } |
| - start := provision.OfflineStart(id, rec.Region, to, d.addr, time.Now()) |
| - probe := *rec |
| + start := begin(id) |
| + probe := *d.rec |
| if err := start(&probe); err != nil { |
| jsonError(w, http.StatusConflict, offlineConflict(err)) |
| return |
| @@ -708,6 +800,7 @@ func (s *Server) offlineReleaseLock(ctx context.Context, host, id string) { |
| // The verbs. |
| const ( |
| + offlineVerbStart = "start" |
| offlineVerbSeal = "seal" |
| offlineVerbFlip = "flip" |
| offlineVerbOpen = "open" |
| @@ -718,6 +811,8 @@ const ( |
| // wait for a box, not for the operator. |
| func offlineVerbOf(state string) string { |
| switch state { |
| + case provision.OfflineStaging: |
| + return offlineVerbStart |
| case provision.OfflineStopping: |
| return offlineVerbSeal |
| case provision.OfflineSealed: |
| @@ -746,6 +841,17 @@ func offlineOpenMut(id string, now time.Time) func(*provision.Record) error { |
| func (s *Server) offlineStepFor(ctx context.Context, d offlineDoor, mv *provision.MoveState, ev offlineEvidence, now time.Time) (offlineStep, *offlineRefusal) { |
| host, from := d.host, s.offlineName(mv.From) |
| switch mv.State { |
| + case provision.OfflineStaging: |
| + // What start would check of the fleet now; its request checks |
| + // (the class, the claims, the region) are the start door's. |
| + fleet, ok := s.readOfflineFleet(ctx) |
| + if !ok { |
| + return offlineStep{}, &offlineRefusal{status: http.StatusConflict, msg: "the fleet's edges files could not all be read. try again"} |
| + } |
| + if ref := s.offlineSelect(ctx, fleet, host, mv.From, mv.To); ref != nil { |
| + return offlineStep{}, ref |
| + } |
| + return offlineStep{mut: provision.OfflineStartStaged(mv.ID, d.addr, now)}, nil |
| case provision.OfflineStopping: |
| fleet, ok := s.readOfflineFleet(ctx) |
| if !ok { |
| @@ -1072,11 +1178,49 @@ type offlineView struct { |
| CanAbort bool `json:"can_abort,omitempty"` |
| Failed bool `json:"failed,omitempty"` |
| Left []offlineLeftView `json:"left,omitempty"` |
| + Stage *offlineStageView `json:"stage,omitempty"` |
| Reports []offlineReportView `json:"reports,omitempty"` |
| Items []offlineItemView `json:"items,omitempty"` |
| Last *provision.MoveOutcome `json:"last,omitempty"` |
| } |
| +// offlineStageView is the pre-copy's progress, from the latest stager |
| +// report: when its last round ended, the replica as that round listed it, |
| +// how much of it the target's staging prefix holds, and what the round |
| +// copied. Phase failed carries the box's own diagnostic in Error. Stager |
| +// reports are rewritten each moves pass, not every few seconds, so their |
| +// age is shown and never judged. |
| +type offlineStageView struct { |
| + Box string `json:"box"` |
| + Phase string `json:"phase"` |
| + At int64 `json:"at"` |
| + ReplicaObjects int `json:"replica_objects"` |
| + ReplicaBytes int64 `json:"replica_bytes"` |
| + StagedObjects int `json:"staged_objects"` |
| + StagedBytes int64 `json:"staged_bytes"` |
| + RoundObjects int `json:"round_objects"` |
| + RoundBytes int64 `json:"round_bytes"` |
| + Error string `json:"error,omitempty"` |
| +} |
| + |
| +// offlineStageOf is the stage view of the newest stager report, nil when no |
| +// box has staged this move. |
| +func offlineStageOf(ev offlineEvidence) *offlineStageView { |
| + var latest *offlineReport |
| + for _, r := range ev.byRole(moves.OfflineRoleStager) { |
| + if latest == nil || r.At.After(latest.At) { |
| + latest = &r |
| + } |
| + } |
| + if latest == nil { |
| + return nil |
| + } |
| + r := latest.OfflineReport |
| + return &offlineStageView{Box: r.Box, Phase: r.Phase, At: r.At.Unix(), |
| + ReplicaObjects: r.ReplicaObjects, ReplicaBytes: r.ReplicaBytes, StagedObjects: r.StagedObjects, StagedBytes: r.StagedBytes, |
| + RoundObjects: r.RoundObjects, RoundBytes: r.RoundBytes, Error: r.StageError} |
| +} |
| + |
| // offlineViewOf builds the status view of a record. Evidence it cannot read |
| // leaves the reports out and says so in Waiting; it never fails the answer. |
| func (s *Server) offlineViewOf(ctx context.Context, rec *provision.Record) offlineView { |
| @@ -1094,7 +1238,7 @@ func (s *Server) offlineViewOf(ctx context.Context, rec *provision.Record) offli |
| v.OpenedAt = mv.OpenedAt.Unix() |
| } |
| switch mv.State { |
| - case provision.OfflineStopping, provision.OfflineSealing, provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded: |
| + case provision.OfflineStaging, provision.OfflineStopping, provision.OfflineSealing, provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded: |
| v.CanAbort = true |
| } |
| ev, err := s.readOfflineEvidence(ctx, rec.Host, mv.ID) |
| @@ -1107,6 +1251,7 @@ func (s *Server) offlineViewOf(ctx context.Context, rec *provision.Record) offli |
| v.Reports = append(v.Reports, offlineReportView{OfflineReport: r.OfflineReport, Fresh: offlineFresh(r.OfflineReport, now)}) |
| } |
| v.Items = offlineItems(ev, mv) |
| + v.Stage = offlineStageOf(ev) |
| for _, r := range ev.byRole(moves.OfflineRoleSource) { |
| for _, p := range r.Retired { |
| if p.Action == moves.RetiredLeft { |
diff --git a/internal/console/apiinstancemove_offline_stage_test.go b/internal/console/apiinstancemove_offline_stage_test.go| new file mode 100644 |
| index 000000000..44a33801b |
| --- /dev/null |
| +++ b/internal/console/apiinstancemove_offline_stage_test.go |
| @@ -0,0 +1,183 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "net/http" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// The pre-copy door (offline-staging), in-process against the offline |
| +// doors' console fixture: the boxes' side is played by the reports they |
| +// write, the record moves only through provision's offline transitions. |
| + |
| +func msgO179(to string) string { |
| + return offlineHost + " is staged for " + to + ". abort the move to move it to another region" |
| +} |
| + |
| +func msgO180(box string) string { |
| + return box + " cannot stage an offline move. it runs an older release" |
| +} |
| + |
| +// stage runs the stage door on a record at rest and keeps the id. |
| +func (w *offlineWorld) stage() offlineView { |
| + w.t.Helper() |
| + before := serialBytes(w.t, w.fx.rel) |
| + v := w.ok("/stage", `{"region":"uk"}`, provision.OfflineStaging) |
| + if v.ID == "" { |
| + w.t.Fatalf("stage answered no id: %+v", v) |
| + } |
| + if serialBytes(w.t, w.fx.rel) == before { |
| + w.t.Fatal("stage did not bump the serial") |
| + } |
| + w.id = v.ID |
| + return v |
| +} |
| + |
| +// staged writes the stager's report as the stager writes it after a round. |
| +func (w *offlineWorld) staged(box string, edit func(*moves.OfflineReport)) { |
| + w.t.Helper() |
| + w.report(moves.OfflineRoleStager, box, moves.OfflinePhaseStaged, func(r *moves.OfflineReport) { |
| + r.Held, r.Fenced, r.Quiesced = false, false, false |
| + r.ReplicaObjects, r.ReplicaBytes, r.StagedObjects, r.StagedBytes, r.RoundObjects, r.RoundBytes = 305, 1_580_000_000, 300, 1_570_000_000, 12, 9_000_000 |
| + if edit != nil { |
| + edit(r) |
| + } |
| + }) |
| +} |
| + |
| +func TestOfflineStageThenStartThenSeal(t *testing.T) { |
| + us := offlineBox{Box: "us-box", Placement: "us", Bucket: "data-us"} |
| + w := newOfflineWorld(t, ieOff("ie-box"), ukOff("uk-box"), us) |
| + v := w.stage() |
| + rec := w.record() |
| + if rec.Region != "ie" || rec.Move.State != provision.OfflineStaging || rec.Move.To != "uk" || rec.Move.StagedAt == nil { |
| + t.Fatalf("staged record: %q %+v", rec.Region, rec.Move) |
| + } |
| + if !v.CanAbort || v.Next != offlineVerbStart || v.Stage != nil { |
| + t.Fatalf("staged view: next %q, can abort %v, stage %+v", v.Next, v.CanAbort, v.Stage) |
| + } |
| + if l, ok := w.lock(); !ok || l.Host != offlineHost || l.Op != w.id { |
| + t.Fatalf("lock %+v %v, want this move's", l, ok) |
| + } |
| + |
| + // Again: another round now. The serial moves, the record does not. |
| + raw, _, _ := w.fx.rel.Get(context.Background(), provision.RecordKey("bab", "shop", offlineHost)) |
| + serial := serialBytes(t, w.fx.rel) |
| + if again := w.ok("/stage", `{"region":"uk"}`, provision.OfflineStaging); again.ID != w.id { |
| + t.Fatalf("a second stage answered move %s, want %s", again.ID, w.id) |
| + } |
| + if serialBytes(t, w.fx.rel) == serial { |
| + t.Fatal("a second stage did not bump the serial") |
| + } |
| + if now, _, _ := w.fx.rel.Get(context.Background(), provision.RecordKey("bab", "shop", offlineHost)); string(now) != string(raw) { |
| + t.Fatal("a second stage wrote the record") |
| + } |
| + |
| + // Another region: refused for stage and for start, nothing written. |
| + w.refused(http.MethodPost, "/stage", `{"region":"us"}`, http.StatusConflict, msgO179("uk")) |
| + w.refused(http.MethodPost, "", `{"region":"us"}`, http.StatusConflict, msgO179("uk")) |
| + // The other verbs wait for start. |
| + for _, verb := range []string{"/seal", "/flip", "/open", "/finish"} { |
| + w.refused(http.MethodPost, verb, "", http.StatusConflict, msgO25(provision.OfflineStaging)) |
| + } |
| + |
| + // The status shows the stager's progress. |
| + w.staged("ie-box", nil) |
| + st := w.status() |
| + if st.Stage == nil || st.Stage.Box != "ie-box" || st.Stage.Phase != moves.OfflinePhaseStaged || st.Stage.StagedObjects != 300 || |
| + st.Stage.ReplicaObjects != 305 || st.Stage.StagedBytes != 1_570_000_000 || st.Stage.RoundBytes != 9_000_000 || st.Stage.At == 0 { |
| + t.Fatalf("stage view %+v", st.Stage) |
| + } |
| + |
| + // Start: staging to stopping, the same move and lock, offline from now. |
| + started := w.ok("", `{"region":"uk"}`, provision.OfflineStopping) |
| + if started.ID != w.id { |
| + t.Fatalf("start answered move %s, want the staged %s", started.ID, w.id) |
| + } |
| + rec = w.record() |
| + if !rec.Move.RequestedAt.After(*rec.Move.StagedAt) || rec.Move.RequestedBy == "" { |
| + t.Fatalf("started record %+v", rec.Move) |
| + } |
| + if l, _ := w.lock(); l.Op != w.id { |
| + t.Fatalf("lock %+v after start", l) |
| + } |
| + if st := w.status(); st.Stage == nil || st.Stage.StagedObjects != 300 { |
| + t.Fatalf("the stage's progress is gone from the status after start: %+v", st.Stage) |
| + } |
| + // From here it is the ordinary move: stage and start cannot be posted |
| + // again, and seal follows the stop. |
| + w.refused(http.MethodPost, "/stage", `{"region":"uk"}`, http.StatusConflict, "a move to uk is already "+provision.OfflineStopping) |
| + w.stopped(nil, "ie-box") |
| + w.ok("/seal", "", provision.OfflineSealing) |
| +} |
| + |
| +// The stage door's own rule: every box in both regions can stage. A box |
| +// that does not publish offline-stage-1 would leave a staging record to |
| +// nobody, so the door refuses before anything is written. The start door |
| +// does not ask it. |
| +func TestOfflineStageRefusesABoxThatCannotStage(t *testing.T) { |
| + for _, c := range []struct { |
| + name string |
| + boxes []offlineBox |
| + box string |
| + }{ |
| + {"source", []offlineBox{{Box: "ie-box", Placement: "ie", Bucket: "data-ie", NoStageCap: true}, ukOff("uk-box")}, "ie-box"}, |
| + {"target", []offlineBox{ieOff("ie-box"), {Box: "uk-box", Placement: "uk", Bucket: "data-uk", NoStageCap: true}}, "uk-box"}, |
| + } { |
| + t.Run(c.name, func(t *testing.T) { |
| + w := newOfflineWorld(t, c.boxes...) |
| + w.refused(http.MethodPost, "/stage", `{"region":"uk"}`, http.StatusConflict, msgO180(c.box)) |
| + if rec := w.record(); rec.Move != nil { |
| + t.Fatalf("a refused stage wrote %+v", rec.Move) |
| + } |
| + if _, ok := w.lock(); ok { |
| + t.Fatal("a refused stage took the lock") |
| + } |
| + w.start() |
| + }) |
| + } |
| + // The start door's refusals hold for stage too (one of them here). |
| + w := newOfflineWorld(t, ieOff("ie-box"), offlineBox{Box: "uk-box", Placement: "uk", Bucket: "data-uk", NoCap: true}) |
| + w.refused(http.MethodPost, "/stage", `{"region":"uk"}`, http.StatusConflict, msgO21("uk-box")) |
| +} |
| + |
| +// Abort from staging: reverting, the region untouched; finish waits for |
| +// every source box's reverted (the stager's comes after it has deleted the |
| +// staged copy), then ends a move that was never offline and frees the lock. |
| +func TestOfflineAbortFromStagingDoor(t *testing.T) { |
| + w := newOfflineWorld(t, oneEach...) |
| + w.stage() |
| + w.staged("ie-box", nil) |
| + w.ok("/abort", "", provision.OfflineReverting) |
| + if rec := w.record(); rec.Region != "ie" || rec.Move.State != provision.OfflineReverting { |
| + t.Fatalf("after abort: %q %+v", rec.Region, rec.Move) |
| + } |
| + w.refused(http.MethodPost, "/finish", "", http.StatusConflict, msgO25(provision.OfflineReverting)) |
| + w.report(moves.OfflineRoleSource, "ie-box", moves.OfflinePhaseReverted, nil) |
| + v := w.ok("/finish", "", "none") |
| + if v.Last == nil || v.Last.Outcome != provision.OfflineOutcomeAborted || v.Last.OfflineS != 0 { |
| + t.Fatalf("last = %+v", v.Last) |
| + } |
| + if _, ok := w.lock(); ok { |
| + t.Fatal("the lock outlived the aborted stage") |
| + } |
| + // The instance can be staged again. |
| + w.stage() |
| +} |
| + |
| +// A failed round shows in the status with the box's own diagnostic. |
| +func TestOfflineStageViewShowsAFailedRound(t *testing.T) { |
| + w := newOfflineWorld(t, oneEach...) |
| + w.stage() |
| + w.staged("ie-box", func(r *moves.OfflineReport) { |
| + r.Phase, r.StageError, r.At = moves.OfflinePhaseFailed, "mirror: copy instances/x: boom", time.Now().UTC() |
| + }) |
| + st := w.status() |
| + if st.Stage == nil || st.Stage.Phase != moves.OfflinePhaseFailed || st.Stage.Error != "mirror: copy instances/x: boom" || st.Stage.StagedObjects != 300 { |
| + t.Fatalf("stage view %+v", st.Stage) |
| + } |
| +} |
diff --git a/internal/console/apiinstancemove_offline_test.go b/internal/console/apiinstancemove_offline_test.go| index e588d9da8..caa67eb91 100644 |
| --- a/internal/console/apiinstancemove_offline_test.go |
| +++ b/internal/console/apiinstancemove_offline_test.go |
| @@ -65,7 +65,8 @@ func offlineClaimFor(t *testing.T, fx *consoleFixture, host, box, region string) |
| // snapshot. |
| type offlineBox struct { |
| Box, Placement, Bucket string |
| - NoCap bool // no offline-move-1 |
| + NoCap bool // no offline-move-1 (nor offline-stage-1) |
| + NoStageCap bool // no offline-stage-1 |
| ClaimsOn bool // its status report says claims are on |
| NoMetrics bool // its status report carries no metrics |
| NotSharedPool bool // serves_shared_pool false |
| @@ -142,6 +143,9 @@ func (w *offlineWorld) fleet(boxes ...offlineBox) { |
| caps := []string{"region-ha-1"} |
| if !b.NoCap { |
| caps = append(caps, offlineMoveCapability) |
| + if !b.NoStageCap { |
| + caps = append(caps, offlineStageCapability) |
| + } |
| } |
| hosts := map[string]any{} |
| for _, h := range b.Hosts { |
| @@ -438,6 +442,7 @@ func msgO31(box string) string { |
| var offlineRoutes = []struct{ method, path, body string }{ |
| {http.MethodGet, "", ""}, |
| {http.MethodPost, "", `{"region":"uk"}`}, |
| + {http.MethodPost, "/stage", `{"region":"uk"}`}, |
| {http.MethodPost, "/seal", ""}, |
| {http.MethodPost, "/flip", ""}, |
| {http.MethodPost, "/open", ""}, |
diff --git a/internal/console/authcli.go b/internal/console/authcli.go| index 97fc1c8bb..91366fa13 100644 |
| --- a/internal/console/authcli.go |
| +++ b/internal/console/authcli.go |
| @@ -720,6 +720,7 @@ func (s *Server) registerCLIRoutes() { |
| // and refused with MV-O20 on a console with host claims or Phase 1 on. |
| s.cliRoute("GET /api/cli/apps/{account}/{app}/instances/{host}/move/offline", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceOfflineStatus) |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceOfflineStart) |
| + s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/stage", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceOfflineStage) |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/seal", svcrole.Allow(svcrole.Instance), s.offlineVerbDoor(offlineVerbSeal)) |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/flip", svcrole.Allow(svcrole.Instance), s.offlineVerbDoor(offlineVerbFlip)) |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/open", svcrole.Allow(svcrole.Instance), s.offlineVerbDoor(offlineVerbOpen)) |
diff --git a/internal/console/claim_invariant_fixtures_test.go b/internal/console/claim_invariant_fixtures_test.go| index cd0643edd..b70fbe60b 100644 |
| --- a/internal/console/claim_invariant_fixtures_test.go |
| +++ b/internal/console/claim_invariant_fixtures_test.go |
| @@ -1488,6 +1488,13 @@ var claimFixtures = map[string]claimFixture{ |
| status: http.StatusConflict, |
| msg: claimOfflineMoveUnsupportedMsg, |
| }}, |
| + "route POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/stage": {refused: &claimRefusalFixture{ |
| + apps: claimRefs("bab/titogo"), |
| + setup: func(t *testing.T, cx *claimCase) { cx.movableInstance(t) }, |
| + do: claimRefusedAPI(http.MethodPost, "/api/cli/apps/bab/titogo/instances/acme.go.tito.io/move/offline/stage", `{"region":"uk"}`), |
| + status: http.StatusConflict, |
| + msg: claimOfflineMoveUnsupportedMsg, |
| + }}, |
| "route POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/seal": {refused: &claimRefusalFixture{ |
| apps: claimRefs("bab/titogo"), |
| setup: func(t *testing.T, cx *claimCase) { cx.offlineMove(t, provision.OfflineStopping) }, |
diff --git a/internal/console/svcroutes_test.go b/internal/console/svcroutes_test.go| index 8a98f4ca0..804ba5518 100644 |
| --- a/internal/console/svcroutes_test.go |
| +++ b/internal/console/svcroutes_test.go |
| @@ -535,6 +535,7 @@ var exemptFromLegacyPin = map[string]bool{ |
| // refuses with MV-O20 on a console with host claims or Phase 1 on. |
| "GET /api/cli/apps/{account}/{app}/instances/{host}/move/offline": true, |
| "POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline": true, |
| + "POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/stage": true, |
| "POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/seal": true, |
| "POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/flip": true, |
| "POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/open": true, |
diff --git a/internal/host/edgesfile.go b/internal/host/edgesfile.go| index 150f394a0..e30d70329 100644 |
| --- a/internal/host/edgesfile.go |
| +++ b/internal/host/edgesfile.go |
| @@ -172,7 +172,8 @@ func edgesFileCapabilities(getenv func(string) string) []string { |
| return []string{capRegionHA1} |
| } |
| -// withOfflineCapability is caps with offline-move-1 when capable. The loop |
| +// withOfflineCapability is caps with offline-move-1 and offline-stage-1 when |
| +// capable (a box that can take part in an offline move can stage one). The loop |
| // passes armedOfflineCapable(): the armed moves worker's own answer, the one |
| // the reconcile hooks act on, so a box never publishes the capability |
| // without the hold that makes it safe (a target that published it with no |
| @@ -181,7 +182,7 @@ func withOfflineCapability(caps []string, capable bool) []string { |
| if !capable { |
| return caps |
| } |
| - return append(slices.Clone(caps), capOfflineMove1) |
| + return append(slices.Clone(caps), capOfflineMove1, capOfflineStage1) |
| } |
| // headroomCapabilities is what headroomFor is asked about: region HA's |
diff --git a/internal/host/movesoffline.go b/internal/host/movesoffline.go| index d272036c8..75bc925df 100644 |
| --- a/internal/host/movesoffline.go |
| +++ b/internal/host/movesoffline.go |
| @@ -330,6 +330,14 @@ func (w *movesWorker) offlineRound(ctx context.Context, s release.Store, rec pro |
| } |
| return nil |
| } |
| + if m.State == provision.OfflineStaging { |
| + // The pre-copy (movesoffline_stage.go): nothing holds or fences |
| + // the instance, and only the stager acts. |
| + if source { |
| + return w.offlineStageStep(ctx, s, rec) |
| + } |
| + return nil |
| + } |
| var errs []error |
| if source { |
| errs = append(errs, w.offlineSourceStep(ctx, s, rec)) |
| @@ -1078,6 +1086,10 @@ func (w *movesWorker) offlineSourceRevert(ctx context.Context, s release.Store, |
| if rt != nil && rt.HoldOp == m.ID { |
| return fmt.Errorf("offline move %s: %s is still held by the move", m.ID, host) |
| } |
| + // A staged move's pre-copy is removed before this box says reverted. |
| + if err := w.offlineUnstage(ctx, s, rec); err != nil { |
| + return err |
| + } |
| r, err := w.sourceReport(host, m, rt, false, moves.OfflinePhaseReverted) |
| if err != nil { |
| return err |
diff --git a/internal/host/movesoffline_stage.go b/internal/host/movesoffline_stage.go| new file mode 100644 |
| index 000000000..41f048f82 |
| --- /dev/null |
| +++ b/internal/host/movesoffline_stage.go |
| @@ -0,0 +1,251 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "os" |
| + "strings" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/park" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The pre-copy of an offline move (offline-staging): while the instance |
| +// keeps serving, one source box (the stager) runs mirror rounds of its |
| +// replica into the target's staging prefix, so the seal's own round, with |
| +// the instance offline, copies only what changed since. |
| +// |
| +// A stage round is the seal's round with nothing else: the same source |
| +// (instances/<host>/ in this box's bucket), the same destination |
| +// (moves/<host>/ in the target's data bucket), the same receipts |
| +// (move-receipts/<host>.json in this box's bucket) and the same mirror pair. |
| +// The staging prefix and the receipts are keyed by host, not by move, so |
| +// the seal finds what the stage copied with no hand-over: lease.MirrorPrefix |
| +// skips an object only when its ETag equals the source's or a receipt for |
| +// this exact pair still matches both sides, deletes every staged key the |
| +// source no longer has (litestream's retention between stage and seal), |
| +// and copies everything else. Nothing a stage round left behind is trusted |
| +// beyond that, so a stale prefix is corrected, never sealed as it is. |
| +// |
| +// The round holds nothing, fences nothing and takes no lease: it reads the |
| +// replica (its listing and the copies' reads) and writes only the target's |
| +// staging prefix and this box's receipts object. The lease, lock.json |
| +// included, is never touched (MirrorPrefix skips lock.json on both sides). |
| +// |
| +// Rounds run on every moves pass while the record stays offline-staging (a |
| +// door's serial bump starts one at once), and stop when it leaves staging: |
| +// the pass that runs a round holds the worker's pass lock, so this box's |
| +// stopped report, which the seal door waits for, is written only after the |
| +// last round it ran has ended. |
| + |
| +// capOfflineStage1 is the edges-file capability a box publishes beside |
| +// offline-move-1 when it can stage: run the pre-copy as the stager, and |
| +// clean it up on abort. The console's stage door refuses a region holding a |
| +// box that does not publish it. |
| +const capOfflineStage1 = "offline-stage-1" |
| + |
| +// offlineStagerOf is moves.OfflineStager over the boxes of the record's |
| +// source placement that the console's doors count (offlineCanHold and the |
| +// capability), read from the edges files now: the console's status view |
| +// names the same box. |
| +func offlineStagerOf(ctx context.Context, s release.Store, home, from string) (string, error) { |
| + bodies, err := loadEdgesFileBodies(ctx, s) |
| + if err != nil { |
| + return "", err |
| + } |
| + var boxes []string |
| + for _, body := range bodies { |
| + var t edgesFile |
| + if err := json.Unmarshal(body, &t); err != nil || t.V < 1 || t.Box == "" { |
| + continue |
| + } |
| + if t.Placement != from && !park.PlacementMatches(home, from, park.PlacementRegion(t.Placement)) { |
| + continue |
| + } |
| + if !t.HibernationConfigured || !t.ServesSharedPool || !containsString(t.Capabilities, capOfflineStage1) { |
| + continue |
| + } |
| + boxes = append(boxes, t.Box) |
| + } |
| + return moves.OfflineStager(boxes), nil |
| +} |
| + |
| +// offlineStageStep is the stager's step in offline-staging: one round, then |
| +// the stager report. Any other source box does nothing. |
| +func (w *movesWorker) offlineStageStep(ctx context.Context, s release.Store, rec provision.Record) error { |
| + cfg := w.cfg |
| + m, host := rec.Move, rec.Host |
| + if cfg.Box == "" { |
| + return errors.New("offline move: this box has no label to report under") |
| + } |
| + stager, err := offlineStagerOf(ctx, s, cfg.ParkHomeRegion, m.From) |
| + if err != nil { |
| + return fmt.Errorf("offline move %s: read the fleet's edges files: %w", m.ID, err) |
| + } |
| + if stager != cfg.Box { |
| + return nil |
| + } |
| + prev, _ := w.ownStagerReport(ctx, s, rec) |
| + r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleStager, Phase: moves.OfflinePhaseStaged} |
| + p, err := w.offlineStageRound(ctx, s, rec) |
| + r.At = cfg.now().UTC() |
| + if err != nil { |
| + // The progress stays what the last good round found; the error is a |
| + // box diagnostic the status view shows as it is. |
| + if prev != nil { |
| + p = stageProgress{replicaObjects: prev.ReplicaObjects, replicaBytes: prev.ReplicaBytes, stagedObjects: prev.StagedObjects, stagedBytes: prev.StagedBytes} |
| + } |
| + p.fill(&r) |
| + r.Phase, r.StageError = moves.OfflinePhaseFailed, err.Error() |
| + return errors.Join(fmt.Errorf("offline move %s: staging %s: %w", m.ID, host, err), writeOfflineReport(ctx, s, r)) |
| + } |
| + p.fill(&r) |
| + if p.roundObjects > 0 { |
| + 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) |
| + } |
| + return writeOfflineReport(ctx, s, r) |
| +} |
| + |
| +// stageProgress is what one stage round found. |
| +type stageProgress struct { |
| + replicaObjects, stagedObjects, roundObjects int |
| + replicaBytes, stagedBytes, roundBytes int64 |
| +} |
| + |
| +func (p stageProgress) fill(r *moves.OfflineReport) { |
| + r.ReplicaObjects, r.ReplicaBytes = p.replicaObjects, p.replicaBytes |
| + r.StagedObjects, r.StagedBytes = p.stagedObjects, p.stagedBytes |
| + r.RoundObjects, r.RoundBytes = p.roundObjects, p.roundBytes |
| +} |
| + |
| +// offlineStageRound is one pre-copy round and the listing that measures it. |
| +func (w *movesWorker) offlineStageRound(ctx context.Context, s release.Store, rec provision.Record) (stageProgress, error) { |
| + cfg := w.cfg |
| + host := rec.Host |
| + tgt, err := w.offlineTarget(ctx, s, rec.Move) |
| + if err != nil { |
| + return stageProgress{}, err |
| + } |
| + bucket, loc, _, err := cfg.tombstoneAt(host) |
| + if err != nil { |
| + return stageProgress{}, err |
| + } |
| + // The activator promotes a host's moves/<host>/ in its OWN bucket over |
| + // its replica on every restore (promoteStagedReplica: staging wins). A |
| + // target that publishes this box's own bucket would have the serving |
| + // instance restore from a copy behind its replica, so the round refuses. |
| + if tgt.DataBucket == bucket { |
| + return stageProgress{}, fmt.Errorf("the target publishes this box's own bucket %s; a staged copy there would replace the serving replica on its next wake", bucket) |
| + } |
| + mirror, err := w.offlineMirror(ctx, tgt.DataBucketRegion) |
| + if err != nil { |
| + return stageProgress{}, err |
| + } |
| + srcPrefix := loc + "/" |
| + dstPrefix := activator.MovesPrefix(host) |
| + receipts := lease.Receipts{Bucket: bucket, Key: activator.MoveReceiptsKey(host)} |
| + round, err := lease.MirrorPrefix(ctx, mirror, bucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts) |
| + if err != nil { |
| + return stageProgress{}, err |
| + } |
| + source, err := mirror.List(ctx, bucket, srcPrefix) |
| + if err != nil { |
| + return stageProgress{}, fmt.Errorf("list the replica: %w", err) |
| + } |
| + staged, err := mirror.List(ctx, tgt.DataBucket, dstPrefix) |
| + if err != nil { |
| + return stageProgress{}, fmt.Errorf("list the staged copy: %w", err) |
| + } |
| + p := stageProgress{roundObjects: round.Changed, roundBytes: round.BytesCopied} |
| + for k, o := range source { |
| + rel := strings.TrimPrefix(k, srcPrefix) |
| + if rel == "lock.json" { |
| + continue |
| + } |
| + p.replicaObjects++ |
| + p.replicaBytes += o.Size |
| + if d, ok := staged[dstPrefix+rel]; ok && d.Size == o.Size { |
| + p.stagedObjects++ |
| + p.stagedBytes += o.Size |
| + } |
| + } |
| + return p, nil |
| +} |
| + |
| +// ownStagerReport is this box's stager report of the move, nil when it has |
| +// written none. One that does not read as a report is an error. |
| +func (w *movesWorker) ownStagerReport(ctx context.Context, s release.Store, rec provision.Record) (*moves.OfflineReport, error) { |
| + key := moves.OfflineReportKey(rec.Host, rec.Move.ID, moves.OfflineRoleStager, w.cfg.Box) |
| + if key == "" { |
| + return nil, fmt.Errorf("offline move %s: %s has no stager report key", rec.Move.ID, rec.Host) |
| + } |
| + raw, _, err := s.Get(ctx, key) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return nil, nil |
| + } |
| + if err != nil { |
| + return nil, err |
| + } |
| + var r moves.OfflineReport |
| + if err := json.Unmarshal(raw, &r); err != nil { |
| + return nil, fmt.Errorf("%s: %w", key, err) |
| + } |
| + return &r, nil |
| +} |
| + |
| +// offlineUnstage is the stager's part of an abort: once any box has staged |
| +// for this move, the stager the fleet names now deletes the target's |
| +// staging prefix and the copy receipts, and says so in its stager report |
| +// (phase reverted). The source step calls it before it reports reverted, |
| +// so the finish that ends the abort waits for it. Every part is an ensure, |
| +// and a move that was never staged does nothing at all, so an abort from |
| +// stopping is exactly what it was. The staging prefix is this host's; no |
| +// other move of the host can be in flight while this record carries this |
| +// one. |
| +func (w *movesWorker) offlineUnstage(ctx context.Context, s release.Store, rec provision.Record) error { |
| + cfg := w.cfg |
| + m, host := rec.Move, rec.Host |
| + stagers, err := readOfflineReports(ctx, s, host, m.ID, moves.OfflineRoleStager) |
| + if err != nil { |
| + return fmt.Errorf("offline move %s: read the stager reports: %w", m.ID, err) |
| + } |
| + if len(stagers) == 0 { |
| + return nil |
| + } |
| + stager, err := offlineStagerOf(ctx, s, cfg.ParkHomeRegion, m.From) |
| + if err != nil { |
| + return fmt.Errorf("offline move %s: read the fleet's edges files: %w", m.ID, err) |
| + } |
| + if stager != cfg.Box { |
| + return nil |
| + } |
| + if own, ok := stagers[cfg.Box]; ok && own.Phase == moves.OfflinePhaseReverted { |
| + return nil |
| + } |
| + tgt, err := w.offlineTarget(ctx, s, m) |
| + if err != nil { |
| + return err |
| + } |
| + mirror, err := w.offlineMirror(ctx, tgt.DataBucketRegion) |
| + if err != nil { |
| + return err |
| + } |
| + prefix := activator.MovesPrefix(host) |
| + if prefix == "moves/" || !strings.HasPrefix(prefix, "moves/") { |
| + return fmt.Errorf("offline move %s: no staging prefix for %s", m.ID, host) |
| + } |
| + if err := lease.DeletePrefix(ctx, mirror, tgt.DataBucket, prefix); err != nil { |
| + return fmt.Errorf("offline move %s: delete %s/%s: %w", m.ID, tgt.DataBucket, prefix, err) |
| + } |
| + clearOfflineReceipts(ctx, mirror, cfg.OwnBucket, host) |
| + log.Printf("moves: %s: offline move %s: the staged copy in %s/%s is deleted", host, m.ID, tgt.DataBucket, prefix) |
| + r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleStager, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC()} |
| + return writeOfflineReport(ctx, s, r) |
| +} |
diff --git a/internal/host/movesoffline_stage_test.go b/internal/host/movesoffline_stage_test.go| new file mode 100644 |
| index 000000000..0ff438fab |
| --- /dev/null |
| +++ b/internal/host/movesoffline_stage_test.go |
| @@ -0,0 +1,322 @@ |
| +//go:build linux |
| + |
| +package host |
| + |
| +import ( |
| + "context" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// The pre-copy (offline-staging), through offlineRound over the seal world: |
| +// the S3 wire protocol, real litestream replicas, the real lease package and |
| +// the production edges-file writer. A stage round copies the replica into |
| +// the target's staging prefix with the instance serving; the seal's round |
| +// then copies only what changed since; abort from staging removes it all. |
| + |
| +// stagerEdges publishes b's edges file as an agent with an armed, capable |
| +// moves worker does: the hibernate coordinates and the offline capabilities |
| +// (offline-move-1, offline-stage-1). The stager is chosen from these. |
| +func (sw *sealWorld) stagerEdges(b *offBox) { |
| + sw.t.Helper() |
| + ident := edgesFile{Box: b.name, Placement: b.region, DataBucket: sealIE, DataBucketRegion: sealIERegion, |
| + HibernationConfigured: true, ServesSharedPool: true, Capabilities: withOfflineCapability(nil, b.w.cfg.OfflineCapable)} |
| + if err := edgesFileWrite(context.Background(), sw.s, b.reg, ident, "", &b.last, false); err != nil { |
| + sw.t.Fatal(err) |
| + } |
| +} |
| + |
| +// stagePass is sw.pass with the edges file stagerEdges writes. |
| +func (sw *sealWorld) stagePass(b *offBox) error { |
| + sw.t.Helper() |
| + b.reconcile() |
| + sw.stagerEdges(b) |
| + ctx := context.Background() |
| + refreshOfflineView(ctx, sw.s, b.w.cfg.Offline) |
| + return b.w.offlineRound(ctx, sw.s, sw.record()) |
| +} |
| + |
| +func (sw *sealWorld) mustStagePass(b *offBox) { |
| + sw.t.Helper() |
| + if err := sw.stagePass(b); err != nil { |
| + sw.t.Fatal(err) |
| + } |
| +} |
| + |
| +func (sw *sealWorld) stage() { |
| + sw.move(provision.OfflineStage(offID, "", "uk", "op@example.com", sw.clock())) |
| +} |
| + |
| +// copySpy records the source key of every server-side copy the fake S3 |
| +// performs from now on. |
| +type copySpy struct { |
| + mu sync.Mutex |
| + keys []string |
| +} |
| + |
| +func (sw *sealWorld) spyCopies() *copySpy { |
| + c := ©Spy{} |
| + sw.fake.FailOn = func(op, key string) error { |
| + if op == "copy" { |
| + c.mu.Lock() |
| + c.keys = append(c.keys, key) |
| + c.mu.Unlock() |
| + } |
| + return nil |
| + } |
| + sw.t.Cleanup(func() { sw.fake.FailOn = nil }) |
| + return c |
| +} |
| + |
| +func (c *copySpy) take() []string { |
| + c.mu.Lock() |
| + defer c.mu.Unlock() |
| + out := c.keys |
| + c.keys = nil |
| + return out |
| +} |
| + |
| +// replicaRel is the replica's listing by key relative to its prefix, lock |
| +// aside; stagedRel the staging prefix's the same way. |
| +func (sw *sealWorld) replicaRel() map[string]lease.MirrorObject { |
| + prefix := sealPrefix + "/" + offHost + "/" |
| + out := map[string]lease.MirrorObject{} |
| + for k, o := range sw.list(sealIE, prefix) { |
| + if rel := strings.TrimPrefix(k, prefix); rel != "lock.json" { |
| + out[rel] = o |
| + } |
| + } |
| + return out |
| +} |
| + |
| +func (sw *sealWorld) stagedRel() map[string]lease.MirrorObject { |
| + out := map[string]lease.MirrorObject{} |
| + for k, o := range sw.list(sealUK, activator.MovesPrefix(offHost)) { |
| + out[strings.TrimPrefix(k, activator.MovesPrefix(offHost))] = o |
| + } |
| + return out |
| +} |
| + |
| +func sameSizes(t *testing.T, staged, replica map[string]lease.MirrorObject) { |
| + t.Helper() |
| + if len(staged) != len(replica) { |
| + t.Fatalf("staged %d objects, replica %d", len(staged), len(replica)) |
| + } |
| + for rel, o := range replica { |
| + if s, ok := staged[rel]; !ok || s.Size != o.Size { |
| + t.Fatalf("%s: staged %+v (%v), replica %+v", rel, s, ok, o) |
| + } |
| + } |
| +} |
| + |
| +func TestOfflineStageThenStartThenSealSkipsThePreCopy(t *testing.T) { |
| + sw := newSealWorld(t) |
| + ie := sw.ie |
| + ieB := sw.sourceBox("ie-b") |
| + sw.instance(ie) |
| + ctx := context.Background() |
| + // A stale staged prefix an earlier aborted move left behind: a file |
| + // under a replica key with other bytes, and one the replica never had. |
| + // Neither may survive into the seal. |
| + for rel := range sw.replicaRel() { |
| + sw.fake.Put(sealUK, activator.MovesPrefix(offHost)+rel, []byte("stale bytes of an earlier move")) |
| + break |
| + } |
| + sw.fake.Put(sealUK, activator.MovesPrefix(offHost)+"0000/stale-extra.ltx", []byte("stale")) |
| + owner0, expires0, gen0 := sw.lock() |
| + |
| + sw.stage() |
| + spy := sw.spyCopies() |
| + sw.stagerEdges(ie) // both boxes publish the capability, |
| + sw.mustStagePass(ieB) // and ie-b is not the stager: it does nothing |
| + if _, ok := sw.reportIf(moves.OfflineRoleStager, "ie-b"); ok || len(spy.take()) != 0 { |
| + t.Fatal("a box that is not the stager staged") |
| + } |
| + sw.mustStagePass(ie) |
| + first := spy.take() |
| + replica := sw.replicaRel() |
| + sameSizes(t, sw.stagedRel(), replica) |
| + r := sw.report(moves.OfflineRoleStager, "ie-a") |
| + if r.Phase != moves.OfflinePhaseStaged || r.ReplicaObjects != len(replica) || r.StagedObjects != len(replica) || r.RoundBytes == 0 || r.StagedBytes != r.ReplicaBytes || r.StageError != "" { |
| + t.Fatalf("stager report %+v (replica %d objects)", r, len(replica)) |
| + } |
| + if len(first) != len(replica) { |
| + t.Fatalf("the first round made %d copies, want %d", len(first), len(replica)) |
| + } |
| + |
| + // Staging held, fenced and leased nothing: the row serves, no fence, |
| + // the lease is as it was, and a wake takes it. |
| + if rt := ie.row(offHost); rt == nil || rt.HoldOp != "" { |
| + t.Fatalf("row %+v during staging", rt) |
| + } |
| + if ie.svc.Fenced(offHost) { |
| + t.Fatal("fenced during staging") |
| + } |
| + if owner, expires, gen := sw.lock(); owner != owner0 || !expires.Equal(expires0) || gen != gen0 { |
| + t.Fatalf("the lease changed during staging: %s %s %d, was %s %s %d", owner, expires, gen, owner0, expires0, gen0) |
| + } |
| + h, err := acquireAsAWake(ctx) |
| + if err != nil { |
| + t.Fatalf("a wake during staging: %v", err) |
| + } |
| + h.Release(ctx) |
| + |
| + // The instance keeps writing; a second round copies only what is new. |
| + sw.writeRows(ie, sw.dbPath(ie), 20, 40, true) |
| + before := sw.stagedRel() |
| + sw.advance(time.Second) |
| + sw.mustStagePass(ie) |
| + second := spy.take() |
| + fresh := 0 |
| + for rel, o := range sw.replicaRel() { |
| + if s, ok := before[rel]; !ok || s.Size != o.Size { |
| + fresh++ |
| + } |
| + } |
| + if fresh == 0 || len(second) != fresh { |
| + t.Fatalf("the second round made %d copies, want the %d new objects", len(second), fresh) |
| + } |
| + sameSizes(t, sw.stagedRel(), sw.replicaRel()) |
| + |
| + // A replica file that came and went between rounds (retention): the |
| + // seal's round removes it from staging. |
| + retired := sealPrefix + "/" + offHost + "/scratch/retired.bin" |
| + sw.fake.Put(sealIE, retired, []byte("gone before the seal")) |
| + sw.mustStagePass(ie) |
| + spy.take() |
| + if _, ok := sw.stagedRel()["scratch/retired.bin"]; !ok { |
| + t.Fatal("fixture: the retired file was not staged") |
| + } |
| + sw.fake.BareDelete(sealIE, retired) |
| + |
| + // Start: the stage's objects are in place, and the seal's round copies |
| + // only what changed since the last stage round. |
| + preStaged := sw.stagedRel() |
| + preReplica := sw.replicaRel() |
| + sw.move(provision.OfflineStartStaged(offID, "op@example.com", sw.clock())) |
| + ie.tick() |
| + ieB.tick() |
| + sw.move(provision.OfflineSeal(offID, "ie-a", []string{"ie-a", "ie-b"})) |
| + spy.take() |
| + sw.mustPass(ie) |
| + if st := sw.record().Move.State; st != provision.OfflineSealed { |
| + t.Fatalf("state %s, want sealed", st) |
| + } |
| + sealCopies := spy.take() |
| + if len(sealCopies) >= len(preReplica) { |
| + t.Fatalf("the seal made %d copies of a %d-object replica the stage had copied", len(sealCopies), len(preReplica)) |
| + } |
| + t.Logf("the seal copied %d of %d objects", len(sealCopies), len(sw.replicaRel())) |
| + for _, k := range sealCopies { |
| + rel := strings.TrimPrefix(k, sealPrefix+"/"+offHost+"/") |
| + s, staged := preStaged[rel] |
| + o, listed := preReplica[rel] |
| + if staged && listed && s.Size == o.Size && s.ETag == o.ETag { |
| + t.Fatalf("the seal recopied %s, which the stage had copied", rel) |
| + } |
| + } |
| + seal, _, ok := sw.sealRecord() |
| + if !ok { |
| + t.Fatal("no seal") |
| + } |
| + if _, ok := seal.Staged["scratch/retired.bin"]; ok { |
| + t.Fatal("the seal kept a staged file the replica no longer has") |
| + } |
| + if _, ok := seal.Staged["0000/stale-extra.ltx"]; ok { |
| + t.Fatal("the seal kept the stale prefix's extra file") |
| + } |
| + sameSizes(t, sw.stagedRel(), sw.replicaRel()) |
| + // The staged copy restores to every row written while staging. |
| + restored := t.TempDir() + "/restored.db" |
| + if ok, err := lease.Restore(ctx, restored, sealUK, strings.TrimSuffix(activator.MovesPrefix(offHost), "/"), sealUKRegion); err != nil || !ok { |
| + t.Fatalf("restore the staged copy: %v %v", ok, err) |
| + } |
| + if n := countRows(t, restored); n != 40 { |
| + t.Fatalf("the staged copy holds %d rows, want 40", n) |
| + } |
| +} |
| + |
| +// Abort from staging: the stager deletes the staged copy and the receipts |
| +// before it reports reverted, every other source box reports reverted as |
| +// for any abort, and finish ends a move that was never offline. Nothing the |
| +// move did not write is touched. |
| +func TestOfflineAbortFromStagingRemovesTheStagedCopy(t *testing.T) { |
| + sw := newSealWorld(t) |
| + ie := sw.ie |
| + sw.instance(ie) |
| + sw.stage() |
| + sw.mustStagePass(ie) |
| + if len(sw.stagedRel()) == 0 { |
| + t.Fatal("fixture: nothing staged") |
| + } |
| + // Receipts, as a round over a multipart source leaves them, and |
| + // neighbours the abort must not touch. |
| + sw.fake.Put(sealIE, activator.MoveReceiptsKey(offHost), []byte(`{"v":1}`)) |
| + sw.fake.Put(sealUK, "moves/other.example.com/0000/x.ltx", []byte("x")) |
| + replica := sw.replicaRel() |
| + |
| + sw.abort() |
| + sw.mustStagePass(ie) |
| + if n := len(sw.stagedRel()); n != 0 { |
| + t.Fatalf("%d staged objects left after the abort", n) |
| + } |
| + if _, ok := sw.fake.Current(sealIE, activator.MoveReceiptsKey(offHost)); ok { |
| + t.Fatal("the receipts are still there") |
| + } |
| + if _, ok := sw.fake.Current(sealUK, "moves/other.example.com/0000/x.ltx"); !ok { |
| + t.Fatal("the abort deleted another host's staging") |
| + } |
| + if got := sw.replicaRel(); len(got) != len(replica) { |
| + t.Fatalf("the replica changed: %d objects, was %d", len(got), len(replica)) |
| + } |
| + if r := sw.report(moves.OfflineRoleStager, "ie-a"); r.Phase != moves.OfflinePhaseReverted { |
| + t.Fatalf("stager report %+v", r) |
| + } |
| + if r := sw.report(moves.OfflineRoleSource, "ie-a"); r.Phase != moves.OfflinePhaseReverted { |
| + t.Fatalf("source report %+v", r) |
| + } |
| + // Again: nothing in the buckets changes. |
| + snap := sw.fake.Snapshot() |
| + sw.mustStagePass(ie) |
| + if sw.fake.Snapshot() != snap { |
| + t.Fatal("a second revert pass wrote to the buckets") |
| + } |
| + rec := sw.move(provision.OfflineFinish(offID, sw.clock())) |
| + if rec.Move != nil || rec.LastMove == nil || rec.LastMove.Outcome != provision.OfflineOutcomeAborted || rec.LastMove.OfflineS != 0 { |
| + t.Fatalf("finish: %+v", rec.LastMove) |
| + } |
| + if rt := ie.row(offHost); rt == nil || rt.HoldOp != "" { |
| + t.Fatalf("row %+v after the abort", rt) |
| + } |
| +} |
| + |
| +// A target that publishes the source's own bucket would have the serving |
| +// instance restore from the staged copy (the activator's staging wins), so |
| +// the round refuses and says why in its report. |
| +func TestOfflineStageRefusesTheSourcesOwnBucket(t *testing.T) { |
| + sw := newSealWorld(t) |
| + ie := sw.ie |
| + sw.instance(ie) |
| + var last []byte |
| + if err := edgesFileWrite(context.Background(), sw.s, sw.uk.reg, edgesFile{Box: "uk-box", Placement: "uk", DataBucket: sealIE, DataBucketRegion: sealIERegion}, "", &last, false); err != nil { |
| + t.Fatal(err) |
| + } |
| + sw.stage() |
| + if err := sw.stagePass(ie); err == nil { |
| + t.Fatal("staged into the source's own bucket") |
| + } |
| + if n := len(sw.list(sealIE, "moves/")); n != 0 { |
| + t.Fatalf("%d objects staged into the source's bucket", n) |
| + } |
| + r := sw.report(moves.OfflineRoleStager, "ie-a") |
| + if r.Phase != moves.OfflinePhaseFailed || !strings.Contains(r.StageError, "own bucket") { |
| + t.Fatalf("stager report %+v", r) |
| + } |
| +} |
diff --git a/internal/host/movesoffline_test.go b/internal/host/movesoffline_test.go| index b44a184d9..dcc39c6f8 100644 |
| --- a/internal/host/movesoffline_test.go |
| +++ b/internal/host/movesoffline_test.go |
| @@ -933,11 +933,11 @@ func TestOfflineCapabilityAndAnIncapableBox(t *testing.T) { |
| if !armedOfflineCapable() { |
| t.Fatal("a capable armed worker does not publish") |
| } |
| - if got := withOfflineCapability(edgesFileCapabilities(env()), armedOfflineCapable()); !slices.Equal(got, []string{capOfflineMove1}) { |
| + if got := withOfflineCapability(edgesFileCapabilities(env()), armedOfflineCapable()); !slices.Equal(got, []string{capOfflineMove1, capOfflineStage1}) { |
| t.Fatalf("capable box: %v", got) |
| } |
| base := edgesFileCapabilities(env("CARLOS_REGION_HA", "1")) |
| - if got := withOfflineCapability(base, true); !slices.Equal(got, []string{capRegionHA1, capOfflineMove1}) || !slices.Equal(base, []string{capRegionHA1}) { |
| + if got := withOfflineCapability(base, true); !slices.Equal(got, []string{capRegionHA1, capOfflineMove1, capOfflineStage1}) || !slices.Equal(base, []string{capRegionHA1}) { |
| t.Fatalf("region HA too: %v (base %v)", got, base) |
| } |
| for _, kv := range [][]string{ |
| @@ -952,7 +952,7 @@ func TestOfflineCapabilityAndAnIncapableBox(t *testing.T) { |
| } |
| incapable := wdCap.box("incapable", "", map[string]string{kv[0]: kv[1]}) |
| armedMovesWorker.Store(incapable.w) |
| - if got := withOfflineCapability(edgesFileCapabilities(env(kv...)), armedOfflineCapable()); slices.Contains(got, capOfflineMove1) { |
| + if got := withOfflineCapability(edgesFileCapabilities(env(kv...)), armedOfflineCapable()); slices.Contains(got, capOfflineMove1) || slices.Contains(got, capOfflineStage1) { |
| t.Fatalf("%s=%q publishes: %v", kv[0], kv[1], got) |
| } |
| } |
| @@ -1128,7 +1128,7 @@ func TestOfflineDormantIsMain(t *testing.T) { |
| if json.Unmarshal(b1, &m1) != nil || json.Unmarshal(b2, &m2) != nil { |
| t.Fatal("edges files do not parse") |
| } |
| - if caps, _ := m2["capabilities"].([]any); len(caps) != 1 || caps[0] != capOfflineMove1 { |
| + if caps, _ := m2["capabilities"].([]any); len(caps) != 2 || caps[0] != capOfflineMove1 || caps[1] != capOfflineStage1 { |
| t.Fatalf("capabilities: %v", m2["capabilities"]) |
| } |
| delete(m2, "capabilities") |
diff --git a/internal/moves/offline.go b/internal/moves/offline.go| index ea6f56fd4..967f7bf84 100644 |
| --- a/internal/moves/offline.go |
| +++ b/internal/moves/offline.go |
| @@ -118,11 +118,14 @@ func OfflineSourced() []OfflineItem { |
| } |
| // The roles a report names: a source-region box that stopped the host, the |
| -// one source box that seals, and the target box that lands. |
| +// one source box that seals, the target box that lands, and the one source |
| +// box that pre-copies the replica while the move is offline-staging (the |
| +// stager: the lexically first box of the source region, OfflineStager). |
| const ( |
| OfflineRoleSource = "source" |
| OfflineRoleSealer = "sealer" |
| OfflineRoleTarget = "target" |
| + OfflineRoleStager = "stager" |
| ) |
| // The phases a report names. |
| @@ -136,6 +139,8 @@ const ( |
| OfflinePhaseLifted = "lifted" |
| OfflinePhaseReverted = "reverted" |
| OfflinePhaseFailed = "failed" |
| + // OfflinePhaseStaged is a stager's report after a pre-copy round. |
| + OfflinePhaseStaged = "staged" |
| ) |
| // OfflineReportVersion is OfflineReport.V. |
| @@ -213,15 +218,33 @@ type OfflineReport struct { |
| // Retired is the parked report's inventory of what the source kept. |
| Retired []RetiredPath `json:"retired,omitempty"` |
| + |
| + // The stager's progress, rewritten after every pre-copy round while |
| + // the move is offline-staging (At is when the round ended): the |
| + // replica as the round's own listing after it found it, lock.json |
| + // aside; how much of it the staged copy holds, key and size for key |
| + // and size; and what the round itself copied. A failed round keeps |
| + // the last progress it had and says why in StageError, a box |
| + // diagnostic and never copy. |
| + ReplicaObjects int `json:"replica_objects,omitempty"` |
| + ReplicaBytes int64 `json:"replica_bytes,omitempty"` |
| + StagedObjects int `json:"staged_objects,omitempty"` |
| + StagedBytes int64 `json:"staged_bytes,omitempty"` |
| + RoundObjects int `json:"round_objects,omitempty"` |
| + RoundBytes int64 `json:"round_bytes,omitempty"` |
| + StageError string `json:"stage_error,omitempty"` |
| } |
| var ( |
| - offlineRoles = map[string]bool{OfflineRoleSource: true, OfflineRoleSealer: true, OfflineRoleTarget: true} |
| + offlineRoles = map[string]bool{OfflineRoleSource: true, OfflineRoleSealer: true, OfflineRoleTarget: true, OfflineRoleStager: true} |
| offlinePhases = map[string]bool{OfflinePhaseStopped: true, OfflinePhaseSealed: true, OfflinePhaseLanded: true, OfflinePhaseOpened: true, |
| - OfflinePhaseParked: true, OfflinePhaseGone: true, OfflinePhaseLifted: true, OfflinePhaseReverted: true, OfflinePhaseFailed: true} |
| + OfflinePhaseParked: true, OfflinePhaseGone: true, OfflinePhaseLifted: true, OfflinePhaseReverted: true, OfflinePhaseFailed: true, |
| + OfflinePhaseStaged: true} |
| ) |
| -// offlineCarries reports whether a role's report may carry the item. |
| +// offlineCarries reports whether a role's report may carry the item. A |
| +// stager's report carries none: the pre-copy decides nothing the doors |
| +// check, and the seal's own round proves the db item. |
| func offlineCarries(role string, it OfflineItem) bool { |
| switch role { |
| case OfflineRoleSource, OfflineRoleSealer: |
| @@ -302,7 +325,8 @@ func offlineSegment(s string) bool { |
| // OfflineReportKey is where a box writes its report: |
| // control/offline-moves/<host>/<op>/<role>/<box>.json. It is "" when host |
| -// fails provision.ValidHostShape, role is not source, sealer or target, or op |
| +// fails provision.ValidHostShape, role is not source, sealer, target or |
| +// stager, or op |
| // or box is not a plain name; a caller that gets "" has a bug and writes |
| // nothing. The prefix is outside every create-only and no-delete prefix, so |
| // a box may rewrite its own report. |
| @@ -371,6 +395,22 @@ func ValidateOfflineSeal(r OfflineReport, host, op string) error { |
| return nil |
| } |
| +// OfflineStager is the box that pre-copies the replica while a move is |
| +// offline-staging: the lexically first of boxes, the source region's boxes |
| +// as the fleet's edges files list them, "" for none. Every source box and |
| +// the console compute it from the same listing, so one box copies; two |
| +// boxes that briefly disagree (a box joined between their listings) only |
| +// copy twice, which the receipts make harmless (lease.Receipts). |
| +func OfflineStager(boxes []string) string { |
| + stager := "" |
| + for _, b := range boxes { |
| + if b != "" && (stager == "" || b < stager) { |
| + stager = b |
| + } |
| + } |
| + return stager |
| +} |
| + |
| // OfflineActiveKey is the one lock of the offline move: while it names a |
| // host whose record still carries that move, no other offline move starts. |
| const OfflineActiveKey = "control/offline-moves/active.json" |
diff --git a/internal/moves/offline_test.go b/internal/moves/offline_test.go| index 50fbc9e13..46e8eca3e 100644 |
| --- a/internal/moves/offline_test.go |
| +++ b/internal/moves/offline_test.go |
| @@ -493,3 +493,33 @@ func TestOfflineSealKeyAndValidation(t *testing.T) { |
| t.Errorf("evidence = %q", ev) |
| } |
| } |
| + |
| +// The stager (offline-staging's pre-copy): its report is a report like any |
| +// other, under its own role, and carries no manifest item; the stager is the |
| +// lexically first box, whatever order the listing gave. |
| +func TestOfflineStagerReport(t *testing.T) { |
| + r := OfflineReport{V: OfflineReportVersion, Op: offTestOp, Host: offTestHost, Box: "box-a", Role: OfflineRoleStager, Phase: OfflinePhaseStaged, |
| + ReplicaObjects: 10, ReplicaBytes: 100, StagedObjects: 9, StagedBytes: 90, RoundObjects: 3, RoundBytes: 30} |
| + if err := ValidateOfflineReport(r); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := OfflineReportKey(offTestHost, offTestOp, OfflineRoleStager, "box-a"); got != "control/offline-moves/"+offTestHost+"/"+offTestOp+"/stager/box-a.json" { |
| + t.Fatalf("key = %q", got) |
| + } |
| + r.Items = map[string]Check{"db": {OK: true}} |
| + if ValidateOfflineReport(r) == nil { |
| + t.Fatal("a stager report carrying an item was accepted") |
| + } |
| + body, err := json.Marshal(OfflineReport{V: 1}) |
| + if err != nil || strings.Contains(string(body), "replica_objects") || strings.Contains(string(body), "stage_error") { |
| + t.Fatalf("the stager's fields are not omitempty: %s", body) |
| + } |
| + for _, c := range []struct { |
| + boxes []string |
| + want string |
| + }{{nil, ""}, {[]string{"box-c", "box-a", "box-b"}, "box-a"}, {[]string{"", "box-z"}, "box-z"}} { |
| + if got := OfflineStager(c.boxes); got != c.want { |
| + t.Errorf("OfflineStager(%v) = %q, want %q", c.boxes, got, c.want) |
| + } |
| + } |
| +} |
diff --git a/internal/provision/offline_move.go b/internal/provision/offline_move.go| index 2af9aa355..dd53aa916 100644 |
| --- a/internal/provision/offline_move.go |
| +++ b/internal/provision/offline_move.go |
| @@ -31,11 +31,19 @@ const MoveKindOffline = "offline" |
| // The offline move's states, in the order a move takes them: |
| // |
| -// offline-stopping -> offline-sealing -> offline-sealed -> offline-landing |
| -// -> offline-landed -> offline-open -> offline-finishing -> (cleared) |
| +// [offline-staging ->] offline-stopping -> offline-sealing -> offline-sealed |
| +// -> offline-landing -> offline-landed -> offline-open -> offline-finishing |
| +// -> (cleared) |
| // |
| -// offline-reverting is abort's state, reached from stopping through landed. |
| +// offline-staging is optional: OfflineStage creates the move there instead |
| +// of in stopping, and the one source box that stages (moves.OfflineStager) |
| +// pre-copies the replica into the target's staging prefix while the |
| +// instance keeps serving. Nothing holds, fences or tombstones the instance |
| +// in staging. OfflineStartStaged then takes it to stopping, and the seal's |
| +// own round finds the pre-copied objects and skips them by their receipts. |
| +// offline-reverting is abort's state, reached from staging through landed. |
| const ( |
| + OfflineStaging = "offline-staging" |
| OfflineStopping = "offline-stopping" |
| OfflineSealing = "offline-sealing" |
| OfflineSealed = "offline-sealed" |
| @@ -118,6 +126,16 @@ func offlineIs(rec *Record, id string) error { |
| // malformed id. It stamps kind, id, from, to, state, requested_by and |
| // requested_at; it does not touch the region. |
| func OfflineStart(id, from, to, by string, now time.Time) func(*Record) error { |
| + return offlineBegin(id, from, to, by, now, OfflineStopping) |
| +} |
| + |
| +// OfflineStage is OfflineStart into offline-staging: the same refusals and |
| +// the same stamps, plus staged_at. The instance keeps serving. |
| +func OfflineStage(id, from, to, by string, now time.Time) func(*Record) error { |
| + return offlineBegin(id, from, to, by, now, OfflineStaging) |
| +} |
| + |
| +func offlineBegin(id, from, to, by string, now time.Time, state string) func(*Record) error { |
| return func(rec *Record) error { |
| if err := seedMoveRefusal(rec); err != nil { |
| return err |
| @@ -145,14 +163,49 @@ func OfflineStart(id, from, to, by string, now time.Time) func(*Record) error { |
| ID: id, |
| From: from, |
| To: to, |
| - State: OfflineStopping, |
| + State: state, |
| RequestedBy: by, |
| RequestedAt: now.UTC(), |
| } |
| + if state == OfflineStaging { |
| + t := now.UTC() |
| + rec.Move.StagedAt = &t |
| + } |
| return nil |
| } |
| } |
| +// OfflineStartStaged is staging to stopping: the operator takes a staged |
| +// instance offline. The record's region must still be the move's source. |
| +// It stamps requested_by and requested_at anew, since the instance is |
| +// offline from now, and requested_at is always after staged_at, which is |
| +// how OfflineFinish tells a move that started from one that never did. |
| +func OfflineStartStaged(id, by string, now time.Time) func(*Record) error { |
| + return func(rec *Record) error { |
| + if err := offlineAt(rec, id, OfflineStaging); err != nil { |
| + return err |
| + } |
| + if rec.Region != rec.Move.From { |
| + return fmt.Errorf("%w: %s", ErrMoveConflict, offlineRegionMoved(rec.Region)) |
| + } |
| + if now.IsZero() { |
| + return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart) |
| + } |
| + t := now.UTC() |
| + if s := rec.Move.StagedAt; s != nil && !t.After(*s) { |
| + t = s.Add(time.Nanosecond) |
| + } |
| + rec.Move.State, rec.Move.RequestedBy, rec.Move.RequestedAt = OfflineStopping, by, t |
| + return nil |
| + } |
| +} |
| + |
| +// startedOffline reports whether the move ever took its instance offline: |
| +// a move created in stopping did at once, a staged one at OfflineStartStaged. |
| +func (m *MoveState) startedOffline() bool { |
| + return m.StagedAt == nil || m.RequestedAt.After(*m.StagedAt) |
| +} |
| + |
| // regionWord is a region as a sentence names it: the home placement's empty |
| // code reads "home". |
| func regionWord(r string) string { |
| @@ -269,8 +322,8 @@ func OfflineFinishingTransition(id string) func(*Record) error { |
| } |
| } |
| -// OfflineAbort puts the move on its way back: any of stopping, sealing, |
| -// sealed, landing or landed becomes reverting. From landing and landed the |
| +// OfflineAbort puts the move on its way back: any of staging, stopping, |
| +// sealing, sealed, landing or landed becomes reverting. From landing and landed the |
| // region goes back to the source in the same write (the flip is undone with |
| // the state, never separately). From open and finishing it refuses, because |
| // the target already serves. From reverting it changes nothing and writes |
| @@ -285,7 +338,7 @@ func OfflineAbort(id string) func(*Record) error { |
| switch rec.Move.State { |
| case OfflineReverting: |
| return errRecordUnchanged |
| - case OfflineStopping, OfflineSealing, OfflineSealed: |
| + case OfflineStaging, OfflineStopping, OfflineSealing, OfflineSealed: |
| case OfflineLanding, OfflineLanded: |
| rec.Region = rec.Move.From |
| default: |
| @@ -300,7 +353,8 @@ func OfflineAbort(id string) func(*Record) error { |
| // legal from finishing (outcome done, region must be the target's) and from |
| // reverting (outcome aborted, region must be the source's); anything else is |
| // refused. The outcome's offline time runs from the request to the open for a |
| -// done move and to now for an aborted one. |
| +// done move and to now for an aborted one; a move aborted while it was still |
| +// staging was never offline, and records none. |
| func OfflineFinish(id string, now time.Time) func(*Record) error { |
| return func(rec *Record) error { |
| if err := offlineIs(rec, id); err != nil { |
| @@ -326,7 +380,7 @@ func OfflineFinish(id string, now time.Time) func(*Record) error { |
| default: |
| return fmt.Errorf("%w: %s", ErrMoveConflict, offlineFinishState(m.State)) |
| } |
| - if s := end.Sub(m.RequestedAt).Seconds(); s > 0 { |
| + if s := end.Sub(m.RequestedAt).Seconds(); s > 0 && m.startedOffline() { |
| out.OfflineS = s |
| } |
| rec.LastMove, rec.Move = out, nil |
diff --git a/internal/provision/offline_stage_test.go b/internal/provision/offline_stage_test.go| new file mode 100644 |
| index 000000000..fb21754bc |
| --- /dev/null |
| +++ b/internal/provision/offline_stage_test.go |
| @@ -0,0 +1,178 @@ |
| +package provision |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "reflect" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The pre-copy (offline-staging): OfflineStage creates the move with the |
| +// instance still serving, OfflineStartStaged takes it to stopping, and abort |
| +// from staging ends a move that was never offline. |
| + |
| +func TestOfflineStageThenStartThenTheWholeMove(t *testing.T) { |
| + ctx := context.Background() |
| + now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC) |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := offlineRecord(t, s) |
| + |
| + got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", now)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + m := got.Move |
| + if !m.Offline() || m.State != OfflineStaging || m.ID != offID || m.To != "uk" || got.Region != "" || |
| + m.StagedAt == nil || !m.StagedAt.Equal(now) || !m.RequestedAt.Equal(now) { |
| + t.Fatalf("staged record: %+v region %q", m, got.Region) |
| + } |
| + // Staging is a move in flight: no second move starts beside it. |
| + if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStart(otherOff, "", "uk", "ops@example.com", now)); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("a start beside a staged move: %v", err) |
| + } |
| + if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(otherOff, "", "us", "ops@example.com", now)); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("a stage beside a staged move: %v", err) |
| + } |
| + |
| + start := now.Add(10 * time.Minute) |
| + got, err = SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "other@example.com", start)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if got.Move.State != OfflineStopping || got.Move.RequestedBy != "other@example.com" || !got.Move.RequestedAt.Equal(start) || !got.Move.StagedAt.Equal(now) { |
| + t.Fatalf("started: %+v", got.Move) |
| + } |
| + // The rest of the move is the ordinary path; offline time counts from |
| + // the start, not the stage: start at +10m, open at +30m. |
| + got = offlineWalk(t, s, rec, offlineSteps(now)[1:]) |
| + if got.Move != nil || got.LastMove.Outcome != OfflineOutcomeDone || got.LastMove.OfflineS != 1200 { |
| + t.Fatalf("finished: %+v", got.LastMove) |
| + } |
| +} |
| + |
| +func TestOfflineStartStagedRefusals(t *testing.T) { |
| + ctx := context.Background() |
| + now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC) |
| + |
| + // From every state but staging, and with another id. |
| + steps := offlineSteps(now) |
| + for walked := 1; walked <= 7; walked++ { |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := offlineRecord(t, s) |
| + offlineWalk(t, s, rec, steps[:walked]) |
| + before := offlineGet(t, s, rec) |
| + if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", now)); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("start-staged from %s: %v", before.Move.State, err) |
| + } |
| + if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) { |
| + t.Fatalf("a refused start-staged from %s changed the record", before.Move.State) |
| + } |
| + } |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := offlineRecord(t, s) |
| + if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", now)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(otherOff, "ops@example.com", now)); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("start-staged with another id: %v", err) |
| + } |
| + // A region written under the staged move. |
| + if _, err := updateRecord(ctx, s, rec.Account, rec.App, rec.Host, func(r *Record) error { r.Region = "us"; return nil }); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", now)); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("start-staged over a moved region: %v", err) |
| + } |
| + |
| + // No other step leaves staging: seal through finish all refuse. |
| + s = release.NewDirStore(t.TempDir()) |
| + rec = offlineRecord(t, s) |
| + if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", now)); err != nil { |
| + t.Fatal(err) |
| + } |
| + before := offlineGet(t, s, rec) |
| + for _, st := range steps[1:] { |
| + if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, st.mut()); !errors.Is(err, ErrMoveConflict) { |
| + t.Errorf("%s from staging: %v", st.name, err) |
| + } |
| + } |
| + if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) { |
| + t.Fatal("a refused step changed the staged record") |
| + } |
| +} |
| + |
| +// A start in the same instant as the stage still lands after it, so the |
| +// finish can tell the two apart. |
| +func TestOfflineStartStagedIsAlwaysAfterTheStage(t *testing.T) { |
| + ctx := context.Background() |
| + now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC) |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := offlineRecord(t, s) |
| + if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", now)); err != nil { |
| + t.Fatal(err) |
| + } |
| + got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", now.Add(-time.Second))) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !got.Move.RequestedAt.After(*got.Move.StagedAt) || !got.Move.startedOffline() { |
| + t.Fatalf("start %v is not after the stage %v", got.Move.RequestedAt, got.Move.StagedAt) |
| + } |
| +} |
| + |
| +// Abort from staging: reverting with the region untouched, then a finish |
| +// that records the move aborted and never offline. Aborted after its start, |
| +// a staged move counts its offline time from the start. |
| +func TestOfflineAbortFromStaging(t *testing.T) { |
| + ctx := context.Background() |
| + now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC) |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := offlineRecord(t, s) |
| + if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", now)); err != nil { |
| + t.Fatal(err) |
| + } |
| + got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)) |
| + if err != nil || got.Move.State != OfflineReverting || got.Region != "" { |
| + t.Fatalf("abort from staging: %v %+v", err, got) |
| + } |
| + done, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now.Add(time.Hour))) |
| + if err != nil || done.Move != nil || done.LastMove.Outcome != OfflineOutcomeAborted || done.LastMove.OfflineS != 0 { |
| + t.Fatalf("finish after abort from staging: %v %+v", err, done.LastMove) |
| + } |
| + |
| + s = release.NewDirStore(t.TempDir()) |
| + rec = offlineRecord(t, s) |
| + for _, mut := range []func(*Record) error{ |
| + OfflineStage(offID, "", "uk", "ops@example.com", now), |
| + OfflineStartStaged(offID, "ops@example.com", now.Add(20*time.Minute)), |
| + OfflineAbort(offID), |
| + } { |
| + if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, mut); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + done, err = SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now.Add(30*time.Minute))) |
| + if err != nil || done.LastMove.Outcome != OfflineOutcomeAborted || done.LastMove.OfflineS != 600 { |
| + t.Fatalf("finish after a started staged move: %v %+v", err, done.LastMove) |
| + } |
| +} |
| + |
| +// staged_at is additive like the other offline fields: a staged journal |
| +// round-trips, and every earlier shape still does (TestOfflineFieldsAreAdditiveOnTheWire). |
| +func TestOfflineStagedAtIsAdditiveOnTheWire(t *testing.T) { |
| + const staged = `{"schema":1,"host":"a.go.example.com","slug":"s","app":"app","account":"bab","region":"ie","channel":"stable","created_at":"2026-09-01T08:00:00Z","move":{"id":"mv_abcdefghijklmnopqrstuvwxyz","from":"ie","to":"uk","state":"offline-staging","requested_by":"ops@example.com","requested_at":"2026-10-21T09:00:00Z","updated_at":"2026-10-21T09:00:00Z","kind":"offline","staged_at":"2026-10-21T09:00:00Z"}}` |
| + rec, err := DecodeRecord([]byte(staged)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if rec.Move.State != OfflineStaging || rec.Move.StagedAt == nil || rec.Move.startedOffline() { |
| + t.Fatalf("staged journal: %+v", rec.Move) |
| + } |
| + if out, _ := json.Marshal(rec); string(out) != staged { |
| + t.Errorf("staged journal did not round-trip:\n got %s\nwant %s", out, staged) |
| + } |
| +} |
diff --git a/internal/provision/provision.go b/internal/provision/provision.go| index 30a4bf1b3..420ec20cb 100644 |
| --- a/internal/provision/provision.go |
| +++ b/internal/provision/provision.go |
| @@ -1065,6 +1065,13 @@ type MoveState struct { |
| // Boxes are the source-region boxes that had to confirm the stop, named |
| // at seal. |
| Boxes []string `json:"boxes,omitempty"` |
| + // StagedAt is when an offline move was staged (OfflineStage): created |
| + // in offline-staging, with the instance still serving. RequestedAt is |
| + // then moved to the start, when the instance went offline, so a staged |
| + // move's offline time counts from the start like any other's; and a |
| + // move aborted before its start was never offline (OfflineFinish). |
| + // Additive and omitempty like the three above. |
| + StagedAt *time.Time `json:"staged_at,omitempty"` |
| } |
| // MoveInputs (what the target was checked against, §7.3 T3) is NOT |