amadan

carlos / platform Public

console: the open record is written create-only before the open (adopted when it is this move's, 409 when another's), and the evidence skips it and the return keys

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 07f4d4c84f458fa4c86738aeb2ef4073f30b286d parent dc790bf
3 files changed, +223 −1
  • internal/console/apiinstancemove_offline.go +98 −1
  • internal/console/offlineopen_test.go +98 −0
  • internal/console/offlinereturn.go +27 −0
diff --git a/internal/console/apiinstancemove_offline.go b/internal/console/apiinstancemove_offline.go
index c32c38773..5bc8e0b22 100644
--- a/internal/console/apiinstancemove_offline.go
+++ b/internal/console/apiinstancemove_offline.go
@@ -240,11 +240,17 @@ func (s *Server) readOfflineEvidence(ctx context.Context, host, id string, m *pr
return offlineEvidence{}, err
}
var ev offlineEvidence
- sealKey := moves.OfflineSealKey(host, id)
+ sealKey, openKey := moves.OfflineSealKey(host, id), moves.OpenRecordKey(host, id)
for _, key := range keys {
if key == sealKey {
continue // read below, as the seal and not as a report
}
+ if key == openKey || offlineReturnKey(prefix, key) {
+ // The open record (moves.OpenRecord) and an operator return's
+ // requests and reports (offlinereturn.go) sit under the move's
+ // prefix and are never its reports.
+ continue
+ }
raw, _, err := s.cfg.Releases.Get(ctx, key)
if errors.Is(err, os.ErrNotExist) {
continue
@@ -1481,10 +1487,101 @@ func (s *Server) offlineVerbDoor(verb string) http.HandlerFunc {
writeRefusal(w, ref)
return
}
+ if verb == offlineVerbOpen {
+ // The open record goes first (create-only, or this move's own
+ // adopted), and the open lands at its stamp.
+ open, err := s.offlineOpenRecord(ctx, d.rec, mv, ev, provision.ProposeStamp(d.rec, time.Now()))
+ if errors.Is(err, errOpenRecordInTheWay) {
+ jsonError(w, http.StatusConflict, msgOpenRecordInTheWay)
+ return
+ }
+ if err != nil {
+ s.offlineWriteFailed(w, d, err)
+ return
+ }
+ step.mut = provision.OfflineOpenTransition(mv.ID, open.OpenedAt, provision.Receipt{})
+ }
s.offlineAdvance(w, r, d, step.mut)
}
}
+// msgOpenRecordInTheWay is the open door's refusal when the open record's
+// key holds another move's record (self-serve moves plan 2 Task 8).
+const msgOpenRecordInTheWay = "another move's open record is in the way"
+
+// errOpenRecordInTheWay: the move's open record key holds a record that is
+// not this move's, of this lifetime, complete.
+var errOpenRecordInTheWay = errors.New("console: offline move: another move's open record is in the way")
+
+// offlineOpenRecord writes the move's open record (moves.OpenRecord)
+// create-only at moves.OpenRecordKey before the open transition, with
+// OpenedAt = proposed: the operator open door and the self-serve job
+// alike. A complete record already there for this move and lifetime is
+// adopted (a retry after a crash between the record and the open): its
+// OpenedAt is the stamp the open carries. Anything else at the key is
+// errOpenRecordInTheWay. A read or write that fails is the error.
+func (s *Server) offlineOpenRecord(ctx context.Context, rec *provision.Record, mv *provision.MoveState, ev offlineEvidence, proposed time.Time) (moves.OpenRecord, error) {
+ key := moves.OpenRecordKey(rec.Host, mv.ID)
+ if key == "" {
+ return moves.OpenRecord{}, errOpenRecordInTheWay
+ }
+ adopt := func(raw []byte) (moves.OpenRecord, error) {
+ var o moves.OpenRecord
+ if err := json.Unmarshal(raw, &o); err != nil || !offlineOpenRecordOf(o, rec, mv) {
+ return moves.OpenRecord{}, errOpenRecordInTheWay
+ }
+ return o, nil
+ }
+ raw, _, err := s.cfg.Releases.Get(ctx, key)
+ switch {
+ case err == nil:
+ return adopt(raw)
+ case !errors.Is(err, os.ErrNotExist):
+ return moves.OpenRecord{}, err
+ }
+ o := moves.OpenRecord{MoveID: mv.ID, Host: rec.Host, Account: rec.Account, App: rec.App, Incarnation: rec.Incarnation,
+ From: mv.From, To: mv.To, StageBucket: mv.StageBucket, StageRegion: mv.StageRegion, Participants: mv.Participants,
+ Layout: offlineOpenLayout(ev, mv), OpenedAt: proposed.UTC()}
+ body, err := json.Marshal(o)
+ if err != nil {
+ return moves.OpenRecord{}, err
+ }
+ if _, err := s.cfg.Releases.Put(ctx, key, body, ""); err != nil {
+ if !errors.Is(err, release.ErrCASConflict) {
+ return moves.OpenRecord{}, err
+ }
+ // Created first by another writer: adopted only when it is this
+ // move's.
+ raw, _, rerr := s.cfg.Releases.Get(ctx, key)
+ if rerr != nil {
+ return moves.OpenRecord{}, rerr
+ }
+ return adopt(raw)
+ }
+ return o, nil
+}
+
+// offlineOpenRecordOf reports whether o is the complete open record of the
+// move mv of rec's lifetime.
+func offlineOpenRecordOf(o moves.OpenRecord, rec *provision.Record, mv *provision.MoveState) bool {
+ return o.MoveID == mv.ID && o.Host == rec.Host && o.Account == rec.Account && o.App == rec.App &&
+ o.Incarnation == rec.Incarnation && o.From == mv.From && o.To == mv.To && !o.OpenedAt.IsZero()
+}
+
+// offlineOpenLayout is the replica's layout the open record names: the
+// sealer's stopped report's, else any source's, else flat.
+func offlineOpenLayout(ev offlineEvidence, mv *provision.MoveState) string {
+ if rep, ok := ev.get(moves.OfflineRoleSource, mv.Sealer); ok && rep.Layout != "" {
+ return rep.Layout
+ }
+ for _, rep := range ev.byRole(moves.OfflineRoleSource) {
+ if rep.Layout != "" {
+ return rep.Layout
+ }
+ }
+ return moves.LayoutFlat
+}
+
func (s *Server) handleAPIInstanceOfflineAbort(w http.ResponseWriter, r *http.Request) {
d, ok := s.offlineGate(w, r)
if !ok {
diff --git a/internal/console/offlineopen_test.go b/internal/console/offlineopen_test.go
new file mode 100644
index 000000000..7aa4929a8
--- /dev/null
+++ b/internal/console/offlineopen_test.go
@@ -0,0 +1,98 @@
+package console
+
+import (
+ "context"
+ "encoding/json"
+ "net/http"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// TestOfflineOpenDoorWritesTheOpenRecordFirst: the operator open door
+// writes the move's open record create-only before the open transition,
+// and the open lands at its OpenedAt; a record of this move already there
+// (a retry after a crash between the two) is adopted, its stamp the open's;
+// another move's record at the key is the 409, and nothing is written.
+func TestOfflineOpenDoorWritesTheOpenRecordFirst(t *testing.T) {
+ ctx := context.Background()
+ readOpen := func(t *testing.T, w *offlineWorld) (moves.OpenRecord, bool) {
+ t.Helper()
+ raw, _, err := w.fx.rel.Get(ctx, moves.OpenRecordKey(offlineHost, w.id))
+ if err != nil {
+ return moves.OpenRecord{}, false
+ }
+ var o moves.OpenRecord
+ if err := json.Unmarshal(raw, &o); err != nil {
+ t.Fatal(err)
+ }
+ return o, true
+ }
+ t.Run("fresh", func(t *testing.T) {
+ w := newOfflineWorld(t, oneEach...)
+ w.toLanded()
+ w.ok("/open", "", provision.OfflineOpen)
+ o, ok := readOpen(t, w)
+ rec := w.record()
+ switch {
+ case !ok:
+ t.Fatal("no open record")
+ case o.MoveID != w.id || o.Host != offlineHost || o.Incarnation != rec.Incarnation || o.From != "ie" || o.To != "uk" || o.Layout == "":
+ t.Fatalf("open record %+v", o)
+ case rec.Move.OpenedAt == nil || !rec.Move.OpenedAt.Equal(o.OpenedAt):
+ t.Fatalf("opened_at %v, open record %v", rec.Move.OpenedAt, o.OpenedAt)
+ }
+ })
+ t.Run("this move's record is adopted", func(t *testing.T) {
+ w := newOfflineWorld(t, oneEach...)
+ w.toLanded()
+ rec := w.record()
+ at := time.Now().Add(-time.Minute).UTC().Truncate(time.Microsecond)
+ o := moves.OpenRecord{MoveID: w.id, Host: offlineHost, Account: rec.Account, App: rec.App, Incarnation: rec.Incarnation,
+ From: "ie", To: "uk", Layout: moves.LayoutFlat, OpenedAt: at}
+ body, _ := json.Marshal(o)
+ offlinePut(t, w.fx.rel, moves.OpenRecordKey(offlineHost, w.id), body)
+ w.ok("/open", "", provision.OfflineOpen)
+ if got := w.record(); got.Move.OpenedAt == nil || !got.Move.OpenedAt.Equal(at) {
+ t.Fatalf("opened_at %v, want the adopted %v", got.Move.OpenedAt, at)
+ }
+ })
+ t.Run("another move's record is in the way", func(t *testing.T) {
+ w := newOfflineWorld(t, oneEach...)
+ w.toLanded()
+ rec := w.record()
+ o := moves.OpenRecord{MoveID: "mv_other", Host: offlineHost, Account: rec.Account, App: rec.App, Incarnation: rec.Incarnation,
+ From: "ie", To: "uk", OpenedAt: time.Now()}
+ body, _ := json.Marshal(o)
+ offlinePut(t, w.fx.rel, moves.OpenRecordKey(offlineHost, w.id), body)
+ before := w.record().RecordSeq
+ w.refused(http.MethodPost, "/open", "", http.StatusConflict, "another move's open record is in the way")
+ if got := w.record(); got.Move.State != provision.OfflineLanded || got.RecordSeq != before {
+ t.Fatalf("the record moved: %+v seq %d (was %d)", got.Move, got.RecordSeq, before)
+ }
+ })
+ t.Run("the evidence skips the open record and the return keys", func(t *testing.T) {
+ w := newOfflineWorld(t, oneEach...)
+ w.toLanded()
+ w.ok("/open", "", provision.OfflineOpen)
+ prefix := moves.OfflineReportPrefix(offlineHost, w.id)
+ offlinePut(t, w.fx.rel, prefix+"return/1-c-00000000.json", []byte(`{}`))
+ offlinePut(t, w.fx.rel, prefix+"reports/return-ready-b-1-c-00000000.json", []byte(`{}`))
+ for _, key := range []string{prefix + "open.json", prefix + "return/1-c-00000000.json", prefix + "reports/return-ready-b-1-c-00000000.json"} {
+ if !offlineReturnKey(prefix, key) && key != moves.OpenRecordKey(offlineHost, w.id) {
+ t.Fatalf("%s is not skipped", key)
+ }
+ }
+ ev, err := w.fx.srv.readOfflineEvidence(ctx, offlineHost, w.id, w.record().Move)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, r := range ev.reports {
+ if r.Role == "" {
+ t.Fatalf("a non-report was read as a report: %+v", r)
+ }
+ }
+ })
+}
diff --git a/internal/console/offlinereturn.go b/internal/console/offlinereturn.go
new file mode 100644
index 000000000..cf835ab38
--- /dev/null
+++ b/internal/console/offlinereturn.go
@@ -0,0 +1,27 @@
+package console
+
+// The operator's return (self-serve moves plan 2 Task 8; spec §7.5): the
+// offline move the other way, run with the operator doors, onto the box a
+// completed forward move left its leftovers on. The handshake:
+//
+// 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;
+// 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;
+// 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).
+
+import "strings"
+
+// offlineReturnKey reports whether key, under a move's report prefix, is
+// one of an operator return's requests (return/) or reports (reports/):
+// never one of the move's own reports.
+func offlineReturnKey(prefix, key string) bool {
+ rest, ok := strings.CutPrefix(key, prefix)
+ return ok && (strings.HasPrefix(rest, "return/") || strings.HasPrefix(rest, "reports/"))
+}