amadan

carlos / platform Public

console: an overlapping commit or stopped is answered as its retry is (Plan 6 Task 3 fix round 1)

When the guarded transition loses, the door re-reads the record once. A
commit that finds this seed committed answers 200 and writes nothing; a
stopped that finds it past stopped answers SD-N54 with the state, as a
sequential second stopped does. Any other loss is SD-N60.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude ce9704cd92193ddda94a0ab60ec99d9bde9814af parent 425ffed
2 files changed, +125 −4
  • internal/console/apiinstanceimport.go +55 −4
  • internal/console/apiinstanceimport_test.go +70 −0

CI passed — run details

diff --git a/internal/console/apiinstanceimport.go b/internal/console/apiinstanceimport.go
index a9fe9fea3..44e2d6936 100644
--- a/internal/console/apiinstanceimport.go
+++ b/internal/console/apiinstanceimport.go
@@ -282,7 +282,7 @@ func (s *Server) importGate(w http.ResponseWriter, r *http.Request) (principal s
return "", nil, "", "", nil, false
}
host = r.PathValue("host")
- notInstance := fmt.Sprintf("%s is not an instance of %s", host, app)
+ notInstance := notAnInstanceOf(host, app)
if !provision.ValidHostShape(host) {
jsonError(w, http.StatusNotFound, notInstance)
return "", nil, "", "", nil, false
@@ -374,7 +374,7 @@ func (s *Server) handleAPIImportUpload(w http.ResponseWriter, r *http.Request) {
switch {
case errors.Is(err, os.ErrNotExist):
releaseScratch()
- jsonError(w, http.StatusNotFound, fmt.Sprintf("%s is not an instance of %s", host, app))
+ jsonError(w, http.StatusNotFound, notAnInstanceOf(host, app))
return
case err != nil:
releaseScratch()
@@ -606,6 +606,28 @@ func putSnapshot(ctx context.Context, data release.Store, path, key string) erro
return err
}
+// notAnInstanceOf is the 404 sentence for a record that is not there.
+func notAnInstanceOf(host, app string) string {
+ return fmt.Sprintf("%s is not an instance of %s", host, app)
+}
+
+// lostTransition reports whether err is a transition that found the
+// import elsewhere, or a record write that lost its race twice.
+func lostTransition(err error) bool {
+ return errors.Is(err, provision.ErrSeedConflict) || errors.Is(err, release.ErrCASConflict)
+}
+
+// importNow re-reads host's record once, after a transition lost, and
+// returns its seed when it is still seed id. Nil means the import is gone,
+// is another seed, or could not be read: the caller answers SD-N60.
+func (s *Server) importNow(ctx context.Context, sqid, app, host, id string) *provision.SeedSource {
+ rec, err := s.readImportRecord(ctx, sqid, app, host)
+ if err != nil || rec.Seed == nil || rec.Seed.ID != id {
+ return nil
+ }
+ return rec.Seed
+}
+
// importTransitionFailure answers a synchronous transition of the import's
// record that did not land. A guarded transition that finds the import
// somewhere else (provision.ErrSeedConflict), or a record write that lost
@@ -614,10 +636,10 @@ func putSnapshot(ctx context.Context, data release.Store, path, key string) erro
// Anything else is the retry sentence, with the cause logged.
func (s *Server) importTransitionFailure(w http.ResponseWriter, app, host string, err error) {
switch {
- case errors.Is(err, provision.ErrSeedConflict), errors.Is(err, release.ErrCASConflict):
+ case lostTransition(err):
jsonError(w, http.StatusConflict, fmt.Sprintf("%s changed while this request was made. Read its state with carlos instances list and try again.", host))
case errors.Is(err, os.ErrNotExist):
- jsonError(w, http.StatusNotFound, fmt.Sprintf("%s is not an instance of %s", host, app))
+ jsonError(w, http.StatusNotFound, notAnInstanceOf(host, app))
default:
log.Printf("console: import %s: write the record: %v", host, err)
jsonError(w, http.StatusBadGateway, importFailed)
@@ -650,6 +672,16 @@ func (s *Server) handleAPIImportStopped(w http.ResponseWriter, r *http.Request)
wctx, wcancel := context.WithTimeout(context.WithoutCancel(r.Context()), instanceStoreTimeout)
defer wcancel()
if _, err := provision.SetRecordSeed(wctx, s.cfg.Releases, acct.Sqid, app, host, provision.SeedStopped(rec.Seed.ID)); err != nil {
+ // An overlapping stopped may have won: this import is past stopped
+ // now, and a sequential second stopped is told so (SD-N54). That
+ // is not success: the caller reads 409 naming ready or later as
+ // the stop having landed. Any other loss is SD-N60.
+ if lostTransition(err) {
+ if now := s.importNow(wctx, acct.Sqid, app, host, rec.Seed.ID); now != nil && pastStopped(now.State) {
+ jsonError(w, http.StatusConflict, fmt.Sprintf("%s is %s. Mark it stopped once, after its upload.", host, now.State))
+ return
+ }
+ }
s.importTransitionFailure(w, app, host, err)
return
}
@@ -684,6 +716,15 @@ func (s *Server) handleAPIImportCommit(w http.ResponseWriter, r *http.Request) {
wctx, wcancel := context.WithTimeout(context.WithoutCancel(r.Context()), instanceStoreTimeout)
defer wcancel()
if _, err := provision.SetRecordSeed(wctx, s.cfg.Releases, acct.Sqid, app, host, provision.SeedCommit(rec.Seed.ID)); err != nil {
+ // An overlapping commit may have won, between the gate's read and
+ // this write: the import is committed, and this is a retry like any
+ // other. The winner audited and bumped, so this writes nothing.
+ if lostTransition(err) {
+ if now := s.importNow(wctx, acct.Sqid, app, host, rec.Seed.ID); now != nil && now.State == seeds.StateCommitted {
+ writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateCommitted})
+ return
+ }
+ }
s.importTransitionFailure(w, app, host, err)
return
}
@@ -692,3 +733,13 @@ func (s *Server) handleAPIImportCommit(w http.ResponseWriter, r *http.Request) {
s.bumpSerial(wctx)
writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateCommitted})
}
+
+// pastStopped reports whether an import's state is one it only reaches
+// after a stopped: the box has the stop, or has gone further.
+func pastStopped(state string) bool {
+ switch state {
+ case seeds.StateReady, seeds.StateSwapped, seeds.StatePlaced, seeds.StateCommitted:
+ return true
+ }
+ return false
+}
diff --git a/internal/console/apiinstanceimport_test.go b/internal/console/apiinstanceimport_test.go
index 83b2aa8cb..c31b83727 100644
--- a/internal/console/apiinstanceimport_test.go
+++ b/internal/console/apiinstanceimport_test.go
@@ -1325,3 +1325,73 @@ func TestImportTransitionsWriteFailure(t *testing.T) {
t.Fatalf("retried commit = %d %v", code, out)
}
}
+
+// An overlapping duplicate: another request lands the same transition
+// between this one's read and its write. A commit is then told what a
+// sequential retry is told (200, committed), and writes no audit row and no
+// serial bump. A stopped is told what a sequential second stopped is told
+// (409 naming the state it is in), not that it changed under it. A hook
+// that moves the import somewhere else still gets SD-N60.
+func TestImportOverlappingTransitions(t *testing.T) {
+ changed := func(host string) string {
+ return host + " changed while this request was made. Read its state with carlos instances list and try again."
+ }
+ for _, c := range []struct {
+ name, verb, action, start string
+ hook func(id string) []func(string) func(*provision.Record) error
+ code int
+ msg string // the error, "" for a 200
+ state string // the record's state after
+ }{
+ {"commit after commit", "commit", "import-committed", seeds.StatePlaced,
+ func(string) []func(string) func(*provision.Record) error {
+ return []func(string) func(*provision.Record) error{provision.SeedCommit}
+ }, http.StatusOK, "", seeds.StateCommitted},
+ {"commit after abort", "commit", "import-committed", seeds.StatePlaced,
+ func(string) []func(string) func(*provision.Record) error {
+ return []func(string) func(*provision.Record) error{provision.SeedAbort}
+ }, http.StatusConflict, changed(importHost), seeds.StateAborted},
+ {"stopped after stopped", "stopped", "import-stopped", seeds.StateUploaded,
+ func(string) []func(string) func(*provision.Record) error {
+ return []func(string) func(*provision.Record) error{provision.SeedStopped}
+ }, http.StatusConflict, importHost + " is ready. Mark it stopped once, after its upload.", seeds.StateReady},
+ {"stopped after stopped and swapped", "stopped", "import-stopped", seeds.StateUploaded,
+ func(string) []func(string) func(*provision.Record) error {
+ return []func(string) func(*provision.Record) error{provision.SeedStopped, provision.SeedSwapped}
+ }, http.StatusConflict, importHost + " is swapped. Mark it stopped once, after its upload.", seeds.StateSwapped},
+ {"stopped after abort", "stopped", "import-stopped", seeds.StateUploaded,
+ func(string) []func(string) func(*provision.Record) error {
+ return []func(string) func(*provision.Record) error{provision.SeedAbort}
+ }, http.StatusConflict, changed(importHost), seeds.StateAborted},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ f := newImportFixture(t)
+ id := newSeedID(t)
+ f.seedRecord(t, importHost, "", id, seeds.KindImport, c.start)
+ steps := c.hook(id)
+ f.fx.srv.cfg.Releases = &movesOnWrite{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), hook: func() {
+ for _, step := range steps {
+ if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, "bab", fromApp, importHost, step(id)); err != nil {
+ t.Error(err)
+ }
+ }
+ }}
+ before := serial(t, f.fx)
+ code, out := postImport(t, f, importHost, c.verb)
+ if code != c.code || str(t, out, "error") != c.msg {
+ t.Fatalf("%s = %d %v, want %d %q", c.verb, code, out, c.code, c.msg)
+ }
+ if c.code == http.StatusOK && (len(out) != 1 || out["state"] != c.state) {
+ t.Fatalf("%s answered %v, want {state: %s}", c.verb, out, c.state)
+ }
+ rec, _ := f.record(t, importHost)
+ requireImportState(t, rec, id, c.state, "")
+ if serial(t, f.fx) != before {
+ t.Fatal("the overlapping request bumped the serial")
+ }
+ if rows := importAudit(t, f, c.action); len(rows) != 0 {
+ t.Fatalf("the overlapping request audited %s: %q", c.action, rows)
+ }
+ })
+ }
+}