amadan

carlos / platform Public

provision: one marker helper for the write primitive and the fence, and the updaters' marker wait ends with its context

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 99ffb42047e097b0cc51f6c94689e0af20ab8c27 parent 9991e16
5 files changed, +108 −23
  • internal/console/lifetimes_create.go +2 −2
  • internal/provision/lifetime_migration.go +43 −19
  • internal/provision/provision.go +7 −1
  • internal/provision/record_sequence.go +1 −1
  • internal/provision/record_sequence_test.go +55 −0
diff --git a/internal/console/lifetimes_create.go b/internal/console/lifetimes_create.go
index 7cc418b97..cc1dfce25 100644
--- a/internal/console/lifetimes_create.go
+++ b/internal/console/lifetimes_create.go
@@ -510,8 +510,8 @@ func (s *Server) fenceRecord(ctx context.Context, acct, app, host string, op fen
// admitted under the hostname lock, so that migration is not pending
// and the fence may rewrite its exact bytes (Task 5).
wctx := ctx
- if obs.Rec != nil && obs.Rec.IncarnationMigrating != "" {
- wctx = provision.WithMigration(ctx, obs.Rec.IncarnationMigrating)
+ if marker, err := provision.RecordMarker(obs.Raw); err == nil && marker != "" {
+ wctx = provision.WithMigration(ctx, marker)
}
etag, _, err := provision.WriteRecord(wctx, s.cfg.Releases, key, obs.Raw, obs.Raw, obs.ETag)
if err == nil {
diff --git a/internal/provision/lifetime_migration.go b/internal/provision/lifetime_migration.go
index 30f0fca01..037feae08 100644
--- a/internal/provision/lifetime_migration.go
+++ b/internal/provision/lifetime_migration.go
@@ -40,28 +40,50 @@ func migrationFrom(ctx context.Context) string {
return id
}
-// checkRecordMarker refuses a write over before's fields when they carry a
-// marker other than the one ctx's migration owns. Every spelling of the
-// field counts, since DecodeRecord matches names without case.
-func checkRecordMarker(ctx context.Context, fields map[string]json.RawMessage) error {
- own := migrationFrom(ctx)
- for name, value := range fields {
- if !strings.EqualFold(name, "incarnation_migrating") {
- continue
- }
- var marker *string
- if err := json.Unmarshal(value, &marker); err != nil {
- return ErrRecordSequence
- }
- if marker != nil && *marker != "" && *marker != own {
- return ErrRecordMigrating
- }
+// RecordMarker is the migration marker a record body carries: the value
+// DecodeRecord reads for incarnation_migrating (encoding/json's own rule
+// across every spelling of the name), "" for none. A value that is not a
+// string is ErrRecordSequence. WriteRecord's refusal and every caller that
+// writes under a record's marker (the fence, an adoption) read it here, so
+// they can never disagree about which marker a body carries.
+func RecordMarker(raw []byte) (string, error) {
+ var probe struct {
+ Marker *string `json:"incarnation_migrating"`
+ }
+ if err := json.Unmarshal(raw, &probe); err != nil {
+ return "", ErrRecordSequence
+ }
+ if probe.Marker == nil {
+ return "", nil
+ }
+ return *probe.Marker, nil
+}
+
+// checkRecordMarker refuses a write over before when it carries a marker
+// other than the one ctx's migration owns.
+func checkRecordMarker(ctx context.Context, before []byte) error {
+ marker, err := RecordMarker(before)
+ if err != nil {
+ return err
+ }
+ if marker != "" && marker != migrationFrom(ctx) {
+ return ErrRecordMigrating
}
return nil
}
-// MarkMigrating is raw with incarnation_migrating set to id, every other
-// field kept as read. Write it with WriteRecord against raw.
+// dropSpellings removes every spelling of name from fields.
+func dropSpellings(fields map[string]json.RawMessage, name string) {
+ for k := range fields {
+ if strings.EqualFold(k, name) {
+ delete(fields, k)
+ }
+ }
+}
+
+// MarkMigrating is raw with incarnation_migrating set to id (any earlier
+// marker, in any spelling, replaced), every other field kept as read. Write
+// it with WriteRecord against raw.
func MarkMigrating(raw []byte, id string) ([]byte, error) {
if id == "" {
return nil, errors.New("provision: a migration marker needs an id")
@@ -74,6 +96,7 @@ func MarkMigrating(raw []byte, id string) ([]byte, error) {
if err != nil {
return nil, err
}
+ dropSpellings(fields, "incarnation_migrating")
fields["incarnation_migrating"] = value
return json.Marshal(fields)
}
@@ -86,7 +109,8 @@ func FinishMigration(marked []byte) ([]byte, error) {
if err != nil {
return nil, err
}
- delete(fields, "incarnation_migrating")
+ dropSpellings(fields, "incarnation_migrating")
+ dropSpellings(fields, "incarnation")
fields["incarnation"] = json.RawMessage(strconv.FormatUint(1, 10))
return json.Marshal(fields)
}
diff --git a/internal/provision/provision.go b/internal/provision/provision.go
index a6fbc0e26..8d6e46ef0 100644
--- a/internal/provision/provision.go
+++ b/internal/provision/provision.go
@@ -1042,7 +1042,13 @@ func updateRecordSeeding(ctx context.Context, s release.Store, acct, app, host s
if !time.Now().Before(deadline) {
return rec, err
}
- time.Sleep(poll)
+ t := time.NewTimer(poll)
+ select {
+ case <-ctx.Done():
+ t.Stop()
+ return nil, fmt.Errorf("provision: record %s: waiting for its migration: %w: %w", key, ctx.Err(), err)
+ case <-t.C:
+ }
case errors.Is(err, release.ErrCASConflict) && !retried:
retried = true
default:
diff --git a/internal/provision/record_sequence.go b/internal/provision/record_sequence.go
index b859cfc2e..f8c307f18 100644
--- a/internal/provision/record_sequence.go
+++ b/internal/provision/record_sequence.go
@@ -125,7 +125,7 @@ func WriteRecord(ctx context.Context, s release.Store, key string, before, body
if err != nil {
return "", 0, err
}
- if err := checkRecordMarker(ctx, fields); err != nil {
+ if err := checkRecordMarker(ctx, before); err != nil {
return "", 0, err
}
_, candidate, err := RecordSequence(body)
diff --git a/internal/provision/record_sequence_test.go b/internal/provision/record_sequence_test.go
index b69e952d9..59c019935 100644
--- a/internal/provision/record_sequence_test.go
+++ b/internal/provision/record_sequence_test.go
@@ -161,3 +161,58 @@ func TestClaimRawUpdateRecordSurfacesTheMarker(t *testing.T) {
}
}
}
+
+func TestUpdaterWaitEndsWithItsContext(t *testing.T) {
+ s, raw, _ := markedFixture(t)
+ shortMigrationWait(t, time.Minute)
+ ctx, cancel := context.WithCancel(context.Background())
+ time.AfterFunc(50*time.Millisecond, cancel)
+ start := time.Now()
+ err := SetRecordRegion(ctx, s, "bab", "shop", markHost, "uk")
+ if !errors.Is(err, context.Canceled) || !errors.Is(err, ErrRecordMigrating) {
+ t.Fatalf("SetRecordRegion = %v, want the cancellation, wrapped with the marker", err)
+ }
+ if waited := time.Since(start); waited > 10*time.Second {
+ t.Fatalf("SetRecordRegion waited %v after its context ended", waited)
+ }
+ if now, _, _ := s.Get(context.Background(), RecordKey("bab", "shop", markHost)); !bytes.Equal(now, raw) {
+ t.Fatalf("record = %s, want it untouched", now)
+ }
+}
+
+func TestTheMarkerIsTheOneDecodeRecordReads(t *testing.T) {
+ s, raw, etag := markedFixture(t)
+ ctx := context.Background()
+ key := RecordKey("bab", "shop", markHost)
+ // Add an earlier spelling with another value: DecodeRecord, and so the
+ // judge and the console, read the later m1.
+ two := bytes.Replace(raw, []byte(`{`), []byte(`{"Incarnation_Migrating":"other",`), 1)
+ tag, err := s.Put(ctx, key, two, etag)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got := mustDecode(t, two).IncarnationMigrating; got != "m1" {
+ t.Fatalf("DecodeRecord reads %q, want m1", got)
+ }
+ if got, err := RecordMarker(two); err != nil || got != "m1" {
+ t.Fatalf("RecordMarker = %q %v, want m1", got, err)
+ }
+ if _, _, err := WriteRecord(WithMigration(ctx, "other"), s, key, two, two, tag); !errors.Is(err, ErrRecordMigrating) {
+ t.Fatalf("the other spelling's id = %v, want refused", err)
+ }
+ // Marking and finishing replace every spelling.
+ marked, err := MarkMigrating(two, "m2")
+ if err != nil || bytes.Contains(marked, []byte("Incarnation_Migrating")) || mustDecode(t, marked).IncarnationMigrating != "m2" {
+ t.Fatalf("MarkMigrating = %s %v, want one spelling carrying m2", marked, err)
+ }
+ if _, _, err := WriteRecord(WithMigration(WithRecordSequence(ctx), "m1"), s, key, two, marked, tag); err != nil {
+ t.Fatalf("the marker's own id = %v, want admitted", err)
+ }
+ final, err := FinishMigration(two)
+ if err != nil || bytes.Contains(final, []byte("ncarnation_")) || mustDecode(t, final).Incarnation != 1 {
+ t.Fatalf("FinishMigration = %s %v, want no marker in any spelling", final, err)
+ }
+ if _, err := RecordMarker([]byte(`{"incarnation_migrating":7}`)); !errors.Is(err, ErrRecordSequence) {
+ t.Fatalf("a marker that is not a string = %v, want ErrRecordSequence", err)
+ }
+}