amadan
console, provision: operator doors that clear a self-serve move finalise it; the operator return's request door, its report and return mode at the stage and start doors
POST .../move/offline/return writes a create-only request under the forward's prefix after checking its open record; the stage and start doors with return_of read the current request's report from the selected target and accept exactly the forward's leftovers; OfflineStageReturn and OfflineStartReturn refuse a record of another lifetime inside the write. An operator finish or abort that clears a self-serve move's record gives the row its outcome first and leaves the release to the pass.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
6 files changed,
+729
−20
internal/console/apiinstancemove_offline.go+76 −13internal/console/authcli.go+1 −0internal/console/offlinereturn.go+277 −7internal/console/offlinereturn_test.go+300 −0internal/provision/offline_move.go+35 −0internal/provision/offline_return_test.go+40 −0
diff --git a/internal/console/apiinstancemove_offline.go b/internal/console/apiinstancemove_offline.go| index c076f84bf..160d84085 100644 |
| --- a/internal/console/apiinstancemove_offline.go |
| +++ b/internal/console/apiinstancemove_offline.go |
| @@ -173,6 +173,8 @@ func offlineConflict(err error) string { |
| strings.HasPrefix(msg, "this offline move is ") && strings.HasSuffix(msg, ". it finishes from finishing or reverting"), |
| strings.HasPrefix(msg, "the record's region is ") && strings.HasSuffix(msg, " the instance changed under the move"), |
| strings.HasPrefix(msg, "the instance is in ") && strings.Contains(msg, ", not "), |
| + // A return's guarded write (self-serve moves plan 2 Task 8). |
| + msg == "the instance was recreated since that move", |
| // The move family's own recorded refusals, as phase1Conflict passes |
| // them. |
| msg == "no move in flight", |
| @@ -782,28 +784,44 @@ func (s *Server) pinParticipant(ctx context.Context, f offlineFleet, b edgesFile |
| // offlineBeginChecks is what start and stage check before either writes, |
| // in this order: offlineRecordChecks, the body, offlineRegionChecks with the |
| -// door's stage rules (an operator door pins no identity). It answers the |
| -// selection, 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, stage bool) (offlineSelection, bool) { |
| +// door's stage rules (an operator door pins no identity), and for a return |
| +// (return_of, on a record at rest) the return's checks (offlineReturnChecks: |
| +// the forward's open record and the selected target's report of the |
| +// forward's leftovers). It answers the selection and, for a return, the |
| +// forward's open record, 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, stage bool) (offlineSelection, *moves.OpenRecord, bool) { |
| ctx := r.Context() |
| if ref := s.offlineRecordChecks(ctx, d.acct.Sqid, d.app, d.host, d.rec); ref != nil { |
| writeRefusal(w, ref) |
| - return offlineSelection{}, false |
| + return offlineSelection{}, nil, false |
| } |
| var req struct { |
| - Region string `json:"region"` |
| + Region string `json:"region"` |
| + ReturnOf string `json:"return_of"` |
| } |
| if !readJSON(w, r, &req) { |
| - return offlineSelection{}, false |
| + return offlineSelection{}, nil, false |
| } |
| - sel, ref := s.offlineRegionChecks(ctx, d.acct.Sqid, d.app, d.host, d.rec, req.Region, stage, false, "") |
| + returnOf := "" |
| + if _, staged := offlineStaged(d.rec); !staged { |
| + returnOf = req.ReturnOf |
| + } |
| + sel, ref := s.offlineRegionChecks(ctx, d.acct.Sqid, d.app, d.host, d.rec, req.Region, stage, false, returnOf) |
| + if ref != nil { |
| + writeRefusal(w, ref) |
| + return offlineSelection{}, nil, false |
| + } |
| + if returnOf == "" { |
| + return sel, nil, true |
| + } |
| + open, ref := s.offlineReturnChecks(ctx, d, sel, strings.TrimSpace(req.Region), returnOf) |
| if ref != nil { |
| writeRefusal(w, ref) |
| - return offlineSelection{}, false |
| + return offlineSelection{}, nil, false |
| } |
| - return sel, true |
| + return sel, open, true |
| } |
| // offlineStagedFor is the refusal of a start or a stage that names another |
| @@ -862,7 +880,7 @@ func (s *Server) handleAPIInstanceOfflineStart(w http.ResponseWriter, r *http.Re |
| if !ok { |
| return |
| } |
| - sel, ok := s.offlineBeginChecks(w, r, d, false) |
| + sel, open, ok := s.offlineBeginChecks(w, r, d, false) |
| if !ok { |
| return |
| } |
| @@ -877,6 +895,9 @@ func (s *Server) handleAPIInstanceOfflineStart(w http.ResponseWriter, r *http.Re |
| return |
| } |
| s.offlineBegin(w, r, d, sel, func(id string, at time.Time) func(*provision.Record) error { |
| + if open != nil { |
| + return provision.OfflineStartReturn(id, d.rec.Region, sel.To, d.addr, at, provision.ReturnOf{ID: open.MoveID, Incarnation: open.Incarnation}) |
| + } |
| return provision.OfflineStart(id, d.rec.Region, sel.To, d.addr, at, "") |
| }) |
| } |
| @@ -892,7 +913,7 @@ func (s *Server) handleAPIInstanceOfflineStage(w http.ResponseWriter, r *http.Re |
| if !ok { |
| return |
| } |
| - sel, ok := s.offlineBeginChecks(w, r, d, true) |
| + sel, open, ok := s.offlineBeginChecks(w, r, d, true) |
| if !ok { |
| return |
| } |
| @@ -903,6 +924,9 @@ func (s *Server) handleAPIInstanceOfflineStage(w http.ResponseWriter, r *http.Re |
| return |
| } |
| s.offlineBegin(w, r, d, sel, func(id string, at time.Time) func(*provision.Record) error { |
| + if open != nil { |
| + return provision.OfflineStageReturn(id, d.rec.Region, sel.To, d.addr, sel.Stage, at, provision.ReturnOf{ID: open.MoveID, Incarnation: open.Incarnation}) |
| + } |
| return provision.OfflineStage(id, d.rec.Region, sel.To, d.addr, sel.Stage, at, "") |
| }) |
| } |
| @@ -1084,6 +1108,9 @@ func (s *Server) offlineWrite(w http.ResponseWriter, r *http.Request, d offlineD |
| } |
| s.bumpSerial(ctx) |
| if rec.Move == nil && d.rec.Move != nil { |
| + if d.rec.Move.SelfServe != nil && s.selfServeDoorCleared(d.rec.Move.ID, rec) { |
| + return rec, true |
| + } |
| if err := s.releaseBoxes(context.WithoutCancel(ctx), moves.OfflineKindMove, d.rec.Move.ID); err != nil { |
| log.Printf("console: offline move %s: releasing the reservation of %s: %v", d.host, d.rec.Move.ID, err) |
| } |
| @@ -1091,6 +1118,42 @@ func (s *Server) offlineWrite(w http.ResponseWriter, r *http.Request, d offlineD |
| return rec, true |
| } |
| +// selfServeDoorCleared routes an operator door that cleared a self-serve |
| +// move's record through the finaliser (self-serve moves plan 2 Task 8): the |
| +// row's outcome first, from the record's last_move, and the rest of the |
| +// finaliser (the effects, the release) left to the pass, under its lease, |
| +// on its next tick. A request of the job's that is unconfirmed is voided |
| +// first: the door's write moved the record past the sequence it was bound |
| +// to, so it can never land. It reports whether the row exists (false: no |
| +// row to finalise, and the door releases as for an operator move). |
| +func (s *Server) selfServeDoorCleared(id string, rec *provision.Record) bool { |
| + row, err := s.store.SelfServeMove(id) |
| + if err != nil { |
| + log.Printf("console: self-serve move %s: the row, after an operator cleared the move: %v (the pass finalises it)", id, err) |
| + return true |
| + } |
| + if row == nil { |
| + return false |
| + } |
| + if row.Outcome != "" || rec.LastMove == nil || rec.LastMove.ID != id { |
| + return true |
| + } |
| + if out, conf := selfServeOutstanding(row); out && !conf && rec.RecordSeq > row.Request.ExpectSeq { |
| + if err := s.store.VoidSelfServeRequest(id, row.Request.Nonce); err != nil { |
| + log.Printf("console: self-serve move %s: voiding the %s request: %v (the pass finalises it)", id, row.Request.Kind, err) |
| + return true |
| + } |
| + if row, err = s.store.SelfServeMove(id); err != nil || row == nil { |
| + return true |
| + } |
| + } |
| + outcome := rec.LastMove.Outcome |
| + if err := s.store.FinishSelfServeMove(id, outcome, s.selfServeOutcomeReason(row, outcome), "", time.Now()); err != nil { |
| + log.Printf("console: self-serve move %s: the outcome, after an operator cleared the move: %v (the pass finalises it)", id, err) |
| + } |
| + return true |
| +} |
| + |
| // offlineAdvance applies one transition and answers 202 with the status |
| // view of the record it left. |
| func (s *Server) offlineAdvance(w http.ResponseWriter, r *http.Request, d offlineDoor, mut func(*provision.Record) error) { |
diff --git a/internal/console/authcli.go b/internal/console/authcli.go| index 5a42cc6c8..90cc3f47a 100644 |
| --- a/internal/console/authcli.go |
| +++ b/internal/console/authcli.go |
| @@ -726,6 +726,7 @@ func (s *Server) registerCLIRoutes() { |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/open", svcrole.Allow(svcrole.Instance), s.offlineVerbDoor(offlineVerbOpen)) |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/finish", svcrole.Allow(svcrole.Instance), s.offlineVerbDoor(offlineVerbFinish)) |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/abort", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceOfflineAbort) |
| + s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/move/offline/return", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceOfflineReturn) |
| s.cliRoute("DELETE /api/cli/apps/{account}/{app}/instances/{host}", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceDelete) |
| // The import door (seeded-instances spec §5): the instance role, like |
| // the create door that begins it. |
diff --git a/internal/console/offlinereturn.go b/internal/console/offlinereturn.go| index cf835ab38..cb5fd3b7e 100644 |
| --- a/internal/console/offlinereturn.go |
| +++ b/internal/console/offlinereturn.go |
| @@ -6,17 +6,36 @@ package console |
| // |
| // 1. POST .../move/offline/return {"region", "return_of"} (operator only) |
| // checks the forward's open record against the record and writes a |
| -// return request, create-only, under the forward's prefix; |
| +// return request, create-only, under the forward's prefix |
| +// (moves.ReturnRequestKey); a new call is a new request, and the |
| +// current one sorts last by (requested_at, key); |
| // 2. the moves pass of every box in that region that was a source of the |
| -// forward reads the newest request and writes an immutable |
| -// return-ready report for it, its leftovers listed against the open |
| -// record; |
| +// forward reads the current request and writes an immutable |
| +// return-ready report for it (moves.ReturnReportKey), its leftovers |
| +// listed against the open record; nothing is ever overwritten; |
| // 3. the operator's stage (or start) door with return_of reads the report |
| // of the current request from the return's selected target and |
| -// accepts exactly the forward's leftovers (targetReadyDecision in |
| -// return mode). |
| +// accepts exactly the forward's leftovers (returnReadyDecision), then |
| +// writes OfflineStageReturn or OfflineStartReturn, which refuse a record |
| +// of another lifetime inside the write. |
| -import "strings" |
| +import ( |
| + "context" |
| + "crypto/rand" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "net/http" |
| + "os" |
| + "path" |
| + "strings" |
| + "time" |
| + |
| + datalayout "github.com/carlosframework/platform/internal/layout" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| // offlineReturnKey reports whether key, under a move's report prefix, is |
| // one of an operator return's requests (return/) or reports (reports/): |
| @@ -25,3 +44,254 @@ func offlineReturnKey(prefix, key string) bool { |
| rest, ok := strings.CutPrefix(key, prefix) |
| return ok && (strings.HasPrefix(rest, "return/") || strings.HasPrefix(rest, "reports/")) |
| } |
| + |
| +// The return's sentences (self-serve moves plan 2 Task 8). |
| +func ssNoCompletedMove(id, from, to string) string { |
| + return fmt.Sprintf("no completed move %s from %s to %s for this instance", id, from, to) |
| +} |
| + |
| +const ( |
| + msgReturnRequested = "return requested; the target box will report its leftovers" |
| + msgReturnRecreated = "the instance was recreated since that move" |
| +) |
| + |
| +// offlineReturnFresh is how recent a return-ready report must be for the |
| +// stage or start door to accept it. |
| +const offlineReturnFresh = 10 * time.Minute |
| + |
| +// offlineReturnView is the return door's 202. |
| +type offlineReturnView struct { |
| + Message string `json:"message"` |
| + Key string `json:"key"` |
| + RequestedAt int64 `json:"requested_at"` |
| +} |
| + |
| +// readOpenRecord is the forward move's open record, nil when absent; one |
| +// that does not parse is absent (it names no completed move). |
| +func (s *Server) readOpenRecord(ctx context.Context, host, id string) (*moves.OpenRecord, error) { |
| + key := moves.OpenRecordKey(host, id) |
| + if key == "" { |
| + return nil, nil |
| + } |
| + raw, _, err := s.cfg.Releases.Get(ctx, key) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return nil, nil |
| + } |
| + if err != nil { |
| + return nil, err |
| + } |
| + var o moves.OpenRecord |
| + if err := json.Unmarshal(raw, &o); err != nil || o.MoveID != id || o.Host != host || o.OpenedAt.IsZero() { |
| + return nil, nil |
| + } |
| + return &o, nil |
| +} |
| + |
| +// offlineReturnOpen checks that returnOf is a completed forward move of |
| +// this record's lifetime from the region (code) asked for into the |
| +// record's region, with nothing in flight: its open record, or the 409. |
| +func (s *Server) offlineReturnOpen(ctx context.Context, rec *provision.Record, region, code, returnOf string) (*moves.OpenRecord, *offlineRefusal) { |
| + none := &offlineRefusal{status: http.StatusConflict, msg: ssNoCompletedMove(returnOf, region, s.offlineName(rec.Region))} |
| + o, err := s.readOpenRecord(ctx, rec.Host, returnOf) |
| + switch { |
| + case err != nil: |
| + log.Printf("console: offline move %s: the open record of %s: %v", rec.Host, returnOf, err) |
| + return nil, &offlineRefusal{status: http.StatusBadGateway, msg: "could not read the move's reports. try again", transient: true} |
| + case o == nil || o.To != rec.Region || o.From != code || o.Account != rec.Account || o.App != rec.App: |
| + return nil, none |
| + case o.Incarnation != rec.Incarnation: |
| + return nil, &offlineRefusal{status: http.StatusConflict, msg: msgReturnRecreated} |
| + case rec.Move != nil: |
| + return nil, none |
| + } |
| + return o, nil |
| +} |
| + |
| +// handleAPIInstanceOfflineReturn is step 1 of the handshake. |
| +func (s *Server) handleAPIInstanceOfflineReturn(w http.ResponseWriter, r *http.Request) { |
| + d, ok := s.offlineGate(w, r) |
| + if !ok { |
| + return |
| + } |
| + var req struct { |
| + Region string `json:"region"` |
| + ReturnOf string `json:"return_of"` |
| + } |
| + if !readJSON(w, r, &req) { |
| + return |
| + } |
| + region := strings.TrimSpace(req.Region) |
| + code := s.homeRegionIn(region) |
| + if region == "" || !provision.ValidRegion(code) { |
| + jsonError(w, http.StatusBadRequest, "invalid region") |
| + return |
| + } |
| + ctx := r.Context() |
| + if _, ref := s.offlineReturnOpen(ctx, d.rec, region, code, req.ReturnOf); ref != nil { |
| + writeRefusal(w, ref) |
| + return |
| + } |
| + now := time.Now().UTC() |
| + rq := moves.ReturnRequest{RequestedAt: now, Region: code, Console: s.selfServeConsoleID()} |
| + rq.Key = moves.NewReturnRequestKey(now, rq.Console, rand.Reader) |
| + key := moves.ReturnRequestKey(d.host, req.ReturnOf, rq.Key) |
| + body, err := json.Marshal(rq) |
| + if err != nil || key == "" { |
| + jsonError(w, http.StatusBadGateway, "could not write the move. try again") |
| + return |
| + } |
| + if _, err := s.cfg.Releases.Put(ctx, key, body, ""); err != nil { |
| + log.Printf("console: offline move %s: the return request of %s: %v", d.host, req.ReturnOf, err) |
| + jsonError(w, http.StatusBadGateway, "could not write the move. try again") |
| + return |
| + } |
| + s.bumpSerial(ctx) |
| + writeJSON(w, http.StatusAccepted, offlineReturnView{Message: msgReturnRequested, Key: rq.Key, RequestedAt: now.UnixNano()}) |
| +} |
| + |
| +// readReturnReport is the return-ready report of the current request of |
| +// returnOf from the target box sel names: it must exist, be written within |
| +// offlineReturnFresh, answer that request (its key and time), come from a |
| +// box whose BoxID and InstanceID are the selected target's, and be one a |
| +// forward source participant (or, for a forward without participants, a |
| +// box in its source region) may write. Any of these failing cannot |
| +// determine the target's facts. |
| +func (s *Server) readReturnReport(ctx context.Context, host, returnOf string, o *moves.OpenRecord, sel offlineSelection, now time.Time) (*moves.OfflineReport, *offlineRefusal) { |
| + unknown := &offlineRefusal{status: http.StatusConflict, msg: couldNotDetermine(factTargetFacts, sel.Target.Label)} |
| + prefix := moves.ReturnRequestPrefix(host, returnOf) |
| + keys, err := s.cfg.Releases.List(ctx, prefix) |
| + if err != nil { |
| + log.Printf("console: offline move %s: the return requests of %s: %v", host, returnOf, err) |
| + return nil, &offlineRefusal{status: http.StatusBadGateway, msg: "could not read the move's reports. try again", transient: true} |
| + } |
| + var reqs []moves.ReturnRequest |
| + for _, k := range keys { |
| + raw, _, err := s.cfg.Releases.Get(ctx, k) |
| + if err != nil { |
| + continue |
| + } |
| + if rq, err := moves.ParseReturnRequest(raw, host, returnOf, k); err == nil { |
| + reqs = append(reqs, rq) |
| + } |
| + } |
| + cur, ok := moves.CurrentReturnRequest(reqs) |
| + if !ok { |
| + return nil, unknown |
| + } |
| + raw, _, err := s.cfg.Releases.Get(ctx, moves.ReturnReportKey(host, returnOf, sel.Target.BoxID, cur.Key)) |
| + if err != nil { |
| + return nil, unknown |
| + } |
| + var rep moves.OfflineReport |
| + if err := json.Unmarshal(raw, &rep); err != nil || moves.ValidateOfflineReport(rep) != nil { |
| + return nil, unknown |
| + } |
| + age := now.Sub(rep.At) |
| + switch { |
| + case rep.Host != host || rep.Op != returnOf || rep.Role != moves.OfflineRoleTarget || rep.Phase != moves.OfflinePhaseReady || rep.Ready == nil, |
| + rep.RequestKey != cur.Key || !rep.Request.Equal(cur.RequestedAt), |
| + age > offlineReturnFresh || age < -moves.OfflineRequestSkew, |
| + rep.BoxID != sel.Target.BoxID || rep.Box != sel.Target.Label || (sel.Target.InstanceID != "" && rep.InstanceID != sel.Target.InstanceID), |
| + !moves.ReturnParticipant(*o, rep.BoxID, rep.InstanceID, sel.To): |
| + return nil, unknown |
| + } |
| + return &rep, nil |
| +} |
| + |
| +// returnReadyDecision decides a return-ready report for the return of the |
| +// forward o in return mode: it accepts exactly the forward's leftovers and |
| +// otherwise decides as targetReadyDecision does. |
| +// |
| +// - the lock: present, owned by the forward's tombstone owner, lifted; |
| +// - the instances prefix: every current object older than the forward's |
| +// open (a newer one is a write since: refused); |
| +// - the nested directory and its mount view: absent, or present and |
| +// empty, for a nested forward; absent for a flat one; |
| +// - park directories: absent or the forward's own source/ and sealer/; |
| +// - the programs file: present or absent (it names the forward's |
| +// programs); |
| +// - everything else: the ordinary rules. |
| +func returnReadyDecision(host, label string, o *moves.OpenRecord, r *moves.TargetReadyReport, want targetWant) string { |
| + if r == nil || r.Leftovers == nil || r.Leftovers.ForwardID != o.MoveID { |
| + return couldNotDetermine(factTargetFacts, label) |
| + } |
| + lo := *r.Leftovers |
| + if lo.LockOwner != moves.TombstoneOwner(o.MoveID) || !lo.LockLifted { |
| + owner := lo.LockOwner |
| + if owner == "" { |
| + owner = "none" |
| + } |
| + return ssBucketLocked(host, owner) |
| + } |
| + if !r.InstancesPrefix.Known { |
| + return couldNotDetermine(factInstances, label) |
| + } |
| + if len(lo.NewerObjects) > 0 { |
| + return ssBucketHeld(host) |
| + } |
| + for _, obj := range r.InstancesPrefix.Objects { |
| + if !obj.LastModified.Before(o.OpenedAt) { |
| + return ssBucketHeld(host) |
| + } |
| + } |
| + for _, d := range lo.ParkDirs { |
| + if d != o.MoveID+"/source/" && d != o.MoveID+"/sealer/" { |
| + return msgSelfServeParkDirty |
| + } |
| + } |
| + dd := path.Dir(firstDataPath(host, r.DataPaths)) |
| + nested := map[string]bool{provision.InstanceDir(dd, host): true, datalayout.RawInstanceDir(dd, host): true} |
| + empty := map[string]bool{} |
| + for _, p := range lo.EmptyNested { |
| + empty[p] = true |
| + } |
| + // The leftovers taken out, the rest is the ordinary decision. |
| + c := *r |
| + c.Lock = moves.Fact{Known: true, OK: true} |
| + c.InstancesPrefix = moves.PrefixFacts{Known: true} |
| + c.ParkDir = moves.Fact{Known: true, OK: true} |
| + if c.ProgramsFile.Known { |
| + c.ProgramsFile.OK = true |
| + } |
| + c.DataPaths = map[string]moves.Fact{} |
| + for p, f := range r.DataPaths { |
| + if nested[p] && f.Known && !f.OK { |
| + if o.Layout != moves.LayoutNested || !empty[p] { |
| + return ssDataHeld(host, p) |
| + } |
| + f.OK = true |
| + } |
| + c.DataPaths[p] = f |
| + } |
| + return targetReadyDecision(host, label, &c, 0, true, want) |
| +} |
| + |
| +// firstDataPath is the report's <host>.db path (the data directory's |
| +// marker targetDataPaths reads), /data/<host>.db otherwise. |
| +func firstDataPath(host string, paths map[string]moves.Fact) string { |
| + for p := range paths { |
| + if path.Base(p) == host+".db" { |
| + return p |
| + } |
| + } |
| + return "/data/" + host + ".db" |
| +} |
| + |
| +// offlineReturnChecks is step 3 for the stage and start doors: the forward's |
| +// open record (again: this record, this lifetime, nothing in flight), the |
| +// selected target's return-ready report for the current request, and the |
| +// leftovers accepted. It answers the open record for the guarded write. |
| +func (s *Server) offlineReturnChecks(ctx context.Context, d offlineDoor, sel offlineSelection, region, returnOf string) (*moves.OpenRecord, *offlineRefusal) { |
| + o, ref := s.offlineReturnOpen(ctx, d.rec, region, sel.To, returnOf) |
| + if ref != nil { |
| + return nil, ref |
| + } |
| + rep, ref := s.readReturnReport(ctx, d.host, returnOf, o, sel, time.Now()) |
| + if ref != nil { |
| + return nil, ref |
| + } |
| + if why := returnReadyDecision(d.host, sel.Target.Label, o, rep.Ready, s.selfServeTargetWant()); why != "" { |
| + return nil, &offlineRefusal{status: http.StatusConflict, msg: why} |
| + } |
| + return o, nil |
| +} |
diff --git a/internal/console/offlinereturn_test.go b/internal/console/offlinereturn_test.go| new file mode 100644 |
| index 000000000..8bbb85382 |
| --- /dev/null |
| +++ b/internal/console/offlinereturn_test.go |
| @@ -0,0 +1,300 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "net/http" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// The operator's return handshake at the console (self-serve moves plan 2 |
| +// Task 8), with the boxes' reports played as the box writes them |
| +// (internal/host movesoffline_return.go; the box's real pass is witnessed |
| +// in internal/e2e/selfserve). The instance sits in ie after a completed |
| +// forward from uk (the forward's source was uk-box); the return goes back |
| +// to uk. |
| + |
| +const returnFwd = "mv_forwardforwardforwardforwa" |
| + |
| +type returnWorld struct { |
| + *offlineWorld |
| + open moves.OpenRecord |
| +} |
| + |
| +func newReturnWorld(t *testing.T, edit func(o *moves.OpenRecord)) *returnWorld { |
| + t.Helper() |
| + w := &returnWorld{offlineWorld: newOfflineWorld(t, oneEach...)} |
| + rec := w.record() |
| + w.open = moves.OpenRecord{MoveID: returnFwd, Host: offlineHost, Account: rec.Account, App: rec.App, Incarnation: rec.Incarnation, |
| + From: "uk", To: "ie", Layout: moves.LayoutFlat, OpenedAt: time.Now().Add(-time.Hour).UTC(), |
| + Participants: &provision.Participants{Source: []provision.Participant{{BoxID: "uk-box", Label: "uk-box"}}, Target: provision.Participant{BoxID: "ie-box", Label: "ie-box"}}} |
| + if edit != nil { |
| + edit(&w.open) |
| + } |
| + body, _ := json.Marshal(w.open) |
| + offlinePut(t, w.fx.rel, moves.OpenRecordKey(offlineHost, returnFwd), body) |
| + return w |
| +} |
| + |
| +// request asks for the return and answers the request key. |
| +func (w *returnWorld) request() string { |
| + w.t.Helper() |
| + code, out := w.req(http.MethodPost, "/return", `{"region":"uk","return_of":"`+returnFwd+`"}`) |
| + if code != http.StatusAccepted { |
| + w.t.Fatalf("return = %d %s", code, out) |
| + } |
| + var v offlineReturnView |
| + if err := json.Unmarshal([]byte(out), &v); err != nil || v.Message != "return requested; the target box will report its leftovers" || v.Key == "" { |
| + w.t.Fatalf("return answered %s", out) |
| + } |
| + return v.Key |
| +} |
| + |
| +// requestAt writes a request as the door would, at a chosen time and key. |
| +func (w *returnWorld) requestAt(at time.Time, key string) { |
| + w.t.Helper() |
| + body, _ := json.Marshal(moves.ReturnRequest{RequestedAt: at, Region: "uk", Console: "c", Key: key}) |
| + offlinePut(w.t, w.fx.rel, moves.ReturnRequestKey(offlineHost, returnFwd, key), body) |
| +} |
| + |
| +// leftovers is a clean flat forward's leftovers on uk-box: the lifted |
| +// tombstone, the forward's park directories, old replica objects. |
| +func (w *returnWorld) ready() *moves.TargetReadyReport { |
| + ok := moves.Fact{Known: true, OK: true} |
| + r := &moves.TargetReadyReport{RouteAbsent: ok, TimersAbsent: ok, LiveLink: ok, ProgramsFile: ok, DiskFree: ok, DiskFreeBytes: 10 << 30, |
| + ParkDir: moves.Fact{Known: true, Detail: "holds " + returnFwd}, MovesPrefix: moves.PrefixFacts{Known: true}, |
| + InstancesPrefix: moves.PrefixFacts{Known: true, Objects: map[string]moves.PrefixObject{"0000/a.ltx": {ETag: "e", Size: 9, LastModified: w.open.OpenedAt.Add(-time.Minute)}}}, |
| + Lock: moves.Fact{Known: true, Detail: "instances/" + offlineHost + "/lock.json"}, DataPaths: map[string]moves.Fact{}, |
| + StagePathArmed: ok, Capabilities: []string{offlineMoveCapability, offlineStageCapability}, SecretsKeyFP: testKeyFP, |
| + Leftovers: &moves.ReturnLeftovers{ForwardID: returnFwd, LockOwner: moves.TombstoneOwner(returnFwd), LockLifted: true, |
| + ParkDirs: []string{returnFwd + "/sealer/", returnFwd + "/source/"}, EmptyNested: []string{}, Layout: moves.LayoutFlat}} |
| + for _, p := range targetDataPaths(offlineHost, nil) { |
| + r.DataPaths[p] = ok |
| + } |
| + return r |
| +} |
| + |
| +// report writes box's report for request key, as the box does. |
| +func (w *returnWorld) report(box, boxID, instance, key string, at time.Time, ready *moves.TargetReadyReport) { |
| + w.t.Helper() |
| + var reqAt time.Time |
| + if raw, _, err := w.fx.rel.Get(context.Background(), moves.ReturnRequestKey(offlineHost, returnFwd, key)); err == nil { |
| + var rq moves.ReturnRequest |
| + _ = json.Unmarshal(raw, &rq) |
| + reqAt = rq.RequestedAt |
| + } |
| + rep := moves.OfflineReport{V: moves.OfflineReportVersion, Op: returnFwd, Host: offlineHost, Box: box, Role: moves.OfflineRoleTarget, |
| + Phase: moves.OfflinePhaseReady, At: at, Request: reqAt, RequestKey: key, Ready: ready, BoxID: boxID, InstanceID: instance} |
| + body, _ := json.Marshal(rep) |
| + offlinePut(w.t, w.fx.rel, moves.ReturnReportKey(offlineHost, returnFwd, boxID, key), body) |
| +} |
| + |
| +func (w *returnWorld) stage() (int, string) { |
| + return w.req(http.MethodPost, "/stage", `{"region":"uk","return_of":"`+returnFwd+`"}`) |
| +} |
| + |
| +// TestOperatorReturnHandshakeAtTheConsole: the request door, the report the |
| +// stage and start doors read, and return mode's exact acceptance. |
| +func TestOperatorReturnHandshakeAtTheConsole(t *testing.T) { |
| + noMove := "no completed move " + returnFwd + " from uk to ie for this instance" |
| + t.Run("the request door", func(t *testing.T) { |
| + w := newReturnWorld(t, nil) |
| + for name, c := range map[string]struct{ body, want string }{ |
| + "another region": {`{"region":"ie","return_of":"` + returnFwd + `"}`, "the instance is already in ie"}, |
| + "absent open record": {`{"region":"uk","return_of":"mv_nosuchmovenosuchmovenosuch"}`, "no completed move mv_nosuchmovenosuchmovenosuch from uk to ie for this instance"}, |
| + } { |
| + code, out := w.req(http.MethodPost, "/return", c.body) |
| + if code != http.StatusConflict || (errorText(out) != c.want && name != "another region") { |
| + t.Errorf("%s: %d %s", name, code, out) |
| + } |
| + } |
| + k1, k2 := w.request(), w.request() |
| + if k1 == k2 { |
| + t.Fatalf("two calls made one request %s", k1) |
| + } |
| + keys, _ := w.fx.rel.List(context.Background(), moves.ReturnRequestPrefix(offlineHost, returnFwd)) |
| + if len(keys) != 2 { |
| + t.Fatalf("requests %v", keys) |
| + } |
| + }) |
| + t.Run("an open record of another lifetime", func(t *testing.T) { |
| + w := newReturnWorld(t, func(o *moves.OpenRecord) { o.Incarnation++ }) |
| + w.refused(http.MethodPost, "/return", `{"region":"uk","return_of":"`+returnFwd+`"}`, http.StatusConflict, "the instance was recreated since that move") |
| + w.refused(http.MethodPost, "/stage", `{"region":"uk","return_of":"`+returnFwd+`"}`, http.StatusConflict, "the instance was recreated since that move") |
| + }) |
| + t.Run("a forward that went elsewhere", func(t *testing.T) { |
| + w := newReturnWorld(t, func(o *moves.OpenRecord) { o.To = "uk"; o.From = "" }) |
| + w.refused(http.MethodPost, "/return", `{"region":"uk","return_of":"`+returnFwd+`"}`, http.StatusConflict, noMove) |
| + }) |
| + t.Run("the leftovers accepted, the return staged", func(t *testing.T) { |
| + for name, edit := range map[string]func(w *returnWorld, r *moves.TargetReadyReport){ |
| + "flat, park directories left": nil, |
| + "flat, fully cleaned": func(w *returnWorld, r *moves.TargetReadyReport) { |
| + r.Leftovers.ParkDirs = nil |
| + r.ParkDir = moves.Fact{Known: true, OK: true} |
| + r.InstancesPrefix.Objects = nil |
| + }, |
| + "flat, the sealer's park directory only": func(w *returnWorld, r *moves.TargetReadyReport) { |
| + r.Leftovers.ParkDirs = []string{returnFwd + "/sealer/"} |
| + }, |
| + "programs file present": func(w *returnWorld, r *moves.TargetReadyReport) { |
| + r.ProgramsFile = moves.Fact{Known: true, Detail: "./cmd/shop"} |
| + }, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + w := newReturnWorld(t, nil) |
| + key := w.request() |
| + r := w.ready() |
| + if edit != nil { |
| + edit(w, r) |
| + } |
| + w.report("uk-box", "uk-box", "", key, time.Now(), r) |
| + if code, out := w.stage(); code != http.StatusAccepted { |
| + t.Fatalf("stage = %d %s", code, out) |
| + } |
| + if rec := w.record(); rec.Move == nil || rec.Move.State != provision.OfflineStaging || rec.Move.ReturnOf != returnFwd { |
| + t.Fatalf("after the stage: %+v", rec.Move) |
| + } |
| + }) |
| + } |
| + }) |
| + t.Run("a nested forward's empty directories accepted", func(t *testing.T) { |
| + for _, partial := range []bool{false, true} { |
| + w := newReturnWorld(t, func(o *moves.OpenRecord) { o.Layout = moves.LayoutNested }) |
| + key := w.request() |
| + r := w.ready() |
| + r.Leftovers.Layout = moves.LayoutNested |
| + paths := targetDataPaths(offlineHost, nil) |
| + nested := []string{paths[0], paths[1]} |
| + if partial { |
| + nested = nested[:1] // the mount view removed already |
| + } |
| + for _, p := range nested { |
| + r.DataPaths[p] = moves.Fact{Known: true} |
| + r.Leftovers.EmptyNested = append(r.Leftovers.EmptyNested, p) |
| + } |
| + w.report("uk-box", "uk-box", "", key, time.Now(), r) |
| + if code, out := w.stage(); code != http.StatusAccepted { |
| + t.Fatalf("partial %v: stage = %d %s", partial, code, out) |
| + } |
| + } |
| + }) |
| + t.Run("anything else refused", func(t *testing.T) { |
| + paths := targetDataPaths(offlineHost, nil) |
| + for name, c := range map[string]struct { |
| + edit func(w *returnWorld, r *moves.TargetReadyReport) |
| + want string |
| + }{ |
| + "a newer object": {func(w *returnWorld, r *moves.TargetReadyReport) { |
| + r.InstancesPrefix.Objects["0001/b.ltx"] = moves.PrefixObject{LastModified: w.open.OpenedAt.Add(time.Minute)} |
| + }, "the target bucket already holds objects for " + offlineHost}, |
| + "a live lock": {func(w *returnWorld, r *moves.TargetReadyReport) { r.Leftovers.LockLifted = false }, |
| + "the target bucket holds a lock for " + offlineHost + " (moved:" + returnFwd + ")"}, |
| + "another move's lock": {func(w *returnWorld, r *moves.TargetReadyReport) { r.Leftovers.LockOwner = "moved:mv_other" }, |
| + "the target bucket holds a lock for " + offlineHost + " (moved:mv_other)"}, |
| + "a foreign park directory": {func(w *returnWorld, r *moves.TargetReadyReport) { |
| + r.Leftovers.ParkDirs = append(r.Leftovers.ParkDirs, "mv_other/source/") |
| + }, "the target box's park directory is not clean"}, |
| + "a non-empty nested directory": {func(w *returnWorld, r *moves.TargetReadyReport) { |
| + r.DataPaths[paths[0]] = moves.Fact{Known: true} |
| + }, "the target box already holds data for " + offlineHost + ": " + paths[0]}, |
| + "a nested directory on a flat forward": {func(w *returnWorld, r *moves.TargetReadyReport) { |
| + r.DataPaths[paths[1]] = moves.Fact{Known: true} |
| + r.Leftovers.EmptyNested = []string{paths[1]} |
| + }, "the target box already holds data for " + offlineHost + ": " + paths[1]}, |
| + "timers left": {func(w *returnWorld, r *moves.TargetReadyReport) { r.TimersAbsent.OK = false }, |
| + "the target box has timers for " + offlineHost}, |
| + "leftovers of another move": {func(w *returnWorld, r *moves.TargetReadyReport) { r.Leftovers.ForwardID = "mv_other" }, |
| + "could not determine the target's facts on uk-box"}, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + w := newReturnWorld(t, nil) |
| + key := w.request() |
| + r := w.ready() |
| + c.edit(w, r) |
| + w.report("uk-box", "uk-box", "", key, time.Now(), r) |
| + w.refused(http.MethodPost, "/stage", `{"region":"uk","return_of":"`+returnFwd+`"}`, http.StatusConflict, c.want) |
| + if rec := w.record(); rec.Move != nil { |
| + t.Fatalf("a refused return wrote %+v", rec.Move) |
| + } |
| + }) |
| + } |
| + }) |
| + t.Run("only the current request's report from the selected participant", func(t *testing.T) { |
| + unknown := "could not determine the target's facts on uk-box" |
| + for name, c := range map[string]func(w *returnWorld){ |
| + "no report": func(w *returnWorld) { w.request() }, |
| + "a non-participant box": func(w *returnWorld) { |
| + w.open.Participants.Source[0].BoxID = "uk-old" |
| + body, _ := json.Marshal(w.open) |
| + offlinePut(w.t, w.fx.rel, moves.OpenRecordKey(offlineHost, returnFwd), body) |
| + w.report("uk-box", "uk-box", "", w.request(), time.Now(), w.ready()) |
| + }, |
| + "the right box id, another instance": func(w *returnWorld) { |
| + w.open.Participants.Source[0].InstanceID = "i-uk-1" |
| + body, _ := json.Marshal(w.open) |
| + offlinePut(w.t, w.fx.rel, moves.OpenRecordKey(offlineHost, returnFwd), body) |
| + w.report("uk-box", "uk-box", "i-uk-2", w.request(), time.Now(), w.ready()) |
| + }, |
| + "a forward source that is not the return's target": func(w *returnWorld) { |
| + w.open.Participants.Source = append(w.open.Participants.Source, provision.Participant{BoxID: "uk-two", Label: "uk-two"}) |
| + body, _ := json.Marshal(w.open) |
| + offlinePut(w.t, w.fx.rel, moves.OpenRecordKey(offlineHost, returnFwd), body) |
| + w.report("uk-two", "uk-two", "", w.request(), time.Now(), w.ready()) |
| + }, |
| + "an older request's report": func(w *returnWorld) { |
| + old := w.request() |
| + w.report("uk-box", "uk-box", "", old, time.Now(), w.ready()) |
| + w.request() // a newer request, not answered yet |
| + }, |
| + "a stale report": func(w *returnWorld) { |
| + w.report("uk-box", "uk-box", "", w.request(), time.Now().Add(-11*time.Minute), w.ready()) |
| + }, |
| + "two requests at one time: the key decides": func(w *returnWorld) { |
| + at := time.Now().UTC() |
| + w.requestAt(at, "1-c-aaaaaaaa") |
| + w.requestAt(at, "1-c-bbbbbbbb") |
| + w.report("uk-box", "uk-box", "", "1-c-aaaaaaaa", time.Now(), w.ready()) |
| + }, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + w := newReturnWorld(t, nil) |
| + c(w) |
| + w.refused(http.MethodPost, "/stage", `{"region":"uk","return_of":"`+returnFwd+`"}`, http.StatusConflict, unknown) |
| + }) |
| + } |
| + // The later key of two at one time is the current request. |
| + w := newReturnWorld(t, nil) |
| + at := time.Now().UTC() |
| + w.requestAt(at, "1-c-aaaaaaaa") |
| + w.requestAt(at, "1-c-bbbbbbbb") |
| + w.report("uk-box", "uk-box", "", "1-c-bbbbbbbb", time.Now(), w.ready()) |
| + if code, out := w.stage(); code != http.StatusAccepted { |
| + t.Fatalf("the later key's report: %d %s", code, out) |
| + } |
| + }) |
| + t.Run("a direct start with return_of", func(t *testing.T) { |
| + w := newReturnWorld(t, nil) |
| + w.report("uk-box", "uk-box", "", w.request(), time.Now(), w.ready()) |
| + code, out := w.req(http.MethodPost, "", `{"region":"uk","return_of":"`+returnFwd+`"}`) |
| + if code != http.StatusAccepted { |
| + t.Fatalf("start = %d %s", code, out) |
| + } |
| + if rec := w.record(); rec.Move == nil || rec.Move.State != provision.OfflineStopping || rec.Move.ReturnOf != returnFwd { |
| + t.Fatalf("after the start: %+v", rec.Move) |
| + } |
| + }) |
| + t.Run("a start or stage without return_of is unchanged", func(t *testing.T) { |
| + w := newReturnWorld(t, nil) |
| + if code, out := w.req(http.MethodPost, "/stage", `{"region":"uk"}`); code != http.StatusAccepted { |
| + t.Fatalf("stage = %d %s", code, out) |
| + } |
| + if rec := w.record(); rec.Move == nil || rec.Move.ReturnOf != "" { |
| + t.Fatalf("a plain stage carried a return: %+v", rec.Move) |
| + } |
| + }) |
| +} |
diff --git a/internal/provision/offline_move.go b/internal/provision/offline_move.go| index 9530e1e72..ec9242f76 100644 |
| --- a/internal/provision/offline_move.go |
| +++ b/internal/provision/offline_move.go |
| @@ -249,6 +249,41 @@ func OfflineStart(id, from, to, by string, now time.Time, returnOf string) func( |
| return offlineBegin(id, from, to, by, now, OfflineStopping, returnOf) |
| } |
| +// ReturnOf is a return's forward move, as the operator's return doors |
| +// pass it to the guarded write: its id and the lifetime (incarnation) the |
| +// forward moved, from the forward's open record. |
| +type ReturnOf struct { |
| + ID string |
| + Incarnation uint64 |
| +} |
| + |
| +// refusedReturnRecreated: a return of a move made in another lifetime of |
| +// the instance (self-serve moves plan 2 Task 8). |
| +const refusedReturnRecreated = "the instance was recreated since that move" |
| + |
| +// returnGuard refuses a return whose forward moved another lifetime. |
| +func returnGuard(ret ReturnOf, begin func(*Record) error) func(*Record) error { |
| + return func(rec *Record) error { |
| + if ret.ID == "" || rec.Incarnation != ret.Incarnation { |
| + return fmt.Errorf("%w: %s", ErrMoveConflict, refusedReturnRecreated) |
| + } |
| + return begin(rec) |
| + } |
| +} |
| + |
| +// OfflineStartReturn is OfflineStart of a return of ret.ID, refused with |
| +// ErrMoveConflict when the record is not of ret's lifetime: the guard is in |
| +// the write, after any read the door made. |
| +func OfflineStartReturn(id, from, to, by string, now time.Time, ret ReturnOf) func(*Record) error { |
| + return returnGuard(ret, OfflineStart(id, from, to, by, now, ret.ID)) |
| +} |
| + |
| +// OfflineStageReturn is OfflineStage of a return of ret.ID, with |
| +// OfflineStartReturn's lifetime guard. |
| +func OfflineStageReturn(id, from, to, by string, st StageTarget, now time.Time, ret ReturnOf) func(*Record) error { |
| + return returnGuard(ret, OfflineStage(id, from, to, by, st, now, ret.ID)) |
| +} |
| + |
| // StageTarget is what a stage fixes for the whole move: the one box that |
| // copies (Stager) and where the copy goes (Bucket, the target's data |
| // bucket, in Region). |
diff --git a/internal/provision/offline_return_test.go b/internal/provision/offline_return_test.go| new file mode 100644 |
| index 000000000..acbeea362 |
| --- /dev/null |
| +++ b/internal/provision/offline_return_test.go |
| @@ -0,0 +1,40 @@ |
| +package provision |
| + |
| +import ( |
| + "errors" |
| + "strings" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// TestOfflineReturnGuardsTheLifetime: a return's start and stage carry the |
| +// forward's lifetime into the write and refuse a record of another |
| +// incarnation (the instance recreated since the forward) with the recorded |
| +// sentence; on the forward's lifetime they are OfflineStart and |
| +// OfflineStage with return_of set. |
| +func TestOfflineReturnGuardsTheLifetime(t *testing.T) { |
| + now := time.Date(2026, 10, 23, 9, 0, 0, 0, time.UTC) |
| + st := StageTarget{Stager: "uk-box", Bucket: "data-ie", Region: "eu-west-1"} |
| + for name, mut := range map[string]func(ret ReturnOf) func(*Record) error{ |
| + "start": func(ret ReturnOf) func(*Record) error { |
| + return OfflineStartReturn(offID, "uk", "", "ops@example.com", now, ret) |
| + }, |
| + "stage": func(ret ReturnOf) func(*Record) error { |
| + return OfflineStageReturn(offID, "uk", "", "ops@example.com", st, now, ret) |
| + }, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + rec := &Record{Host: "a.go.example.com", Region: "uk", Incarnation: 4} |
| + err := mut(ReturnOf{ID: otherOff, Incarnation: 3})(rec) |
| + if !errors.Is(err, ErrMoveConflict) || !strings.HasSuffix(err.Error(), "the instance was recreated since that move") || rec.Move != nil { |
| + t.Fatalf("another lifetime: %v, move %+v", err, rec.Move) |
| + } |
| + if err := mut(ReturnOf{Incarnation: 4})(rec); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("no forward id: %v", err) |
| + } |
| + if err := mut(ReturnOf{ID: otherOff, Incarnation: 4})(rec); err != nil || rec.Move == nil || rec.Move.ReturnOf != otherOff { |
| + t.Fatalf("this lifetime: %v, move %+v", err, rec.Move) |
| + } |
| + }) |
| + } |
| +} |