amadan
provision: the lifetime judge observes the record then the slot in bounded rounds; judged readers, listings and updates act on the observation
Co-Authored-By: Codex Astra <noreply@openai.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
9 files changed,
+1240
−38
docs/superpowers/notes/census-proposal-self-serve-moves-1.md+4 −0internal/provision/lifetime.go+12 −3internal/provision/lifetime_judge.go+265 −0internal/provision/lifetime_judge_test.go+760 −0internal/provision/provision.go+67 −20internal/provision/rootrecord.go+25 −12internal/provision/rootrecord_test.go+76 −0internal/provision/self_serve_moves1_census_test.go+27 −2internal/provision/testdata/plan14-root-reads-self-serve-moves-1.tsv+4 −1
diff --git a/docs/superpowers/notes/census-proposal-self-serve-moves-1.md b/docs/superpowers/notes/census-proposal-self-serve-moves-1.md| index 1bfd5616c..b22b0c549 100644 |
| --- a/docs/superpowers/notes/census-proposal-self-serve-moves-1.md |
| +++ b/docs/superpowers/notes/census-proposal-self-serve-moves-1.md |
| @@ -5,3 +5,7 @@ Status: proposed, marked PROPOSED-self-serve-moves-1. Not an accepted ruling; th |
| Task 1 adds four fields to `provision.Record` (`incarnation`, `incarnation_migrating`, `move_seq`, `retired`) and a new file, `internal/provision/lifetime.go`, holding the host incarnation slot, the lifetime verdict, the tombstone body and `GetRecordRaw`. The root-read overlay `internal/provision/testdata/plan14-root-reads-self-serve-moves-1.tsv` re-pins the `internal/provision` source envelope and nothing else: no new root read, class or task, no accepted-legacy row, and the seal stands. The registry schema line is untouched, since `Record` is not a registry struct. The record-write overlay is `plan14SelfServeMoves1WriteCensusInventory` in `internal/provision/self_serve_moves1_census_test.go`, built over the roster-binding inventory: it removes `GetRecordETag` (it no longer builds a record key; it reads through `GetRecordRaw`), adds `GetRecordRaw` as a read-only key-site, and adds `WriteHostIncarnation` as nonrecord (it Puts only `control/host-incarnations/<host>.json`, conditionally). Later tasks of the plan will extend both overlays. |
| Task 2 adds 15 console lifetime-store units and re-pins `migrate`, `migrateAppsKey`, `BackfillOpsMemberships` and `claimSQLCensus`. Each has a proposed legacy disposition limited to console SQL, randomness or an interface declaration, running as `carlos-console`; none adds a root decision. The two constructors `CreateAppInRegion` and `CreateTemporaryAppInRegion` now delegate their SQL to `insertAppRow`: their stale read rows are removed with explicit H3 dispositions, while the declarations remain covered by the source envelope. These account for all six changed frozen declarations, including the gate census changed in fix round 1. The removal parser requires a `census-ruling-` prefix, so those tokens say `census-ruling-PROPOSED-self-serve-moves-1`; they remain proposals. Every unit carries `rev3=PROPOSED-self-serve-moves-1` and a reviewed one-line basis. The console envelope covers the tables, constructors, lifetime projection, negative-read and transition guards, and the table-scoped `app_lifetime` exemption; the provision envelope adds `ErrLifetimeExhausted`. The existing overlay header is retained as the Task 1 record. Its provision coverage digest is refreshed in place because the loader forbids duplicate coverage rows in one overlay. Nineteen new legacy fingerprints require a new pinned seal, checked after verifying the preceding seal. Registry fields and the record-write inventory are unchanged. A green census checks these proposed rows against the code; Paul's ruling is still outstanding. |
| + |
| +Task 6a adds the provision lifetime judge and judged readers, listings and updates. The root census reported the provision envelope and two stale declaration pins: `IndexRecords` and `RecordIndex.Find`. Both retain their authority classification and B3,C ownership. `IndexRecords` now retains every candidate owner key; `Find` judges current bodies on each lookup, requires exactly one valid candidate, refuses corruption and read failures, and validates the chosen record's schema and shape. Nil-judge duplicate candidates remain ambiguous. The overlay records these reviewed dispositions and refreshes the provision envelope over the judge, context helpers, shared record decoding, update guards and slot-counter read limit. No root-read unit is removed, no legacy fingerprint is added, and the legacy seal and registry-field pin stand. |
| + |
| +The write census reported four changed sites and three added sites. `GetRecordRaw` remains a read-only key-site and shares its decoder with `ObserveWith`. `updateRecordSeeding` and `claimRawUpdateRecord` remain record-writers through `WriteRecord`: the former judges before each mutation, including its CAS retry; the latter judges before checking the caller's ETag. `IndexRecords` remains a read-only key-site. `ListRecordsCompleteJudged`, `ListRecordsStrictJudged` and `readAllRecordsJudged` are added as read-only key-sites, with separate reviewed dispositions for their completeness and error behavior. Every listed missing body remains a gap; only a present stale body is silently omitted. All Task 6a entries remain `PROPOSED-self-serve-moves-1`, awaiting Paul's census ruling. The removal loop and host and console consumers remain Task 6b work. |
diff --git a/internal/provision/lifetime.go b/internal/provision/lifetime.go| index 24781bbae..ce79ae718 100644 |
| --- a/internal/provision/lifetime.go |
| +++ b/internal/provision/lifetime.go |
| @@ -5,6 +5,7 @@ import ( |
| "encoding/json" |
| "errors" |
| "fmt" |
| + "math" |
| "os" |
| "strconv" |
| @@ -79,6 +80,9 @@ func ReadHostIncarnation(ctx context.Context, s release.Store, host string) (*Ho |
| if !slot.Valid() { |
| return nil, "", fmt.Errorf("%w: %s: failed validation", ErrSlotMalformed, key) |
| } |
| + if slot.Incarnation > math.MaxInt64 { |
| + return nil, etag, fmt.Errorf("%s: %w", host, ErrLifetimeExhausted) |
| + } |
| return &slot, etag, nil |
| } |
| @@ -171,12 +175,17 @@ func GetRecordRaw(ctx context.Context, s release.Store, acct, app, host string) |
| if err != nil { |
| return nil, nil, "", err |
| } |
| + rec, err = decodeRecordAt(raw, acct, app, host) |
| + return rec, raw, etag, err |
| +} |
| + |
| +func decodeRecordAt(raw []byte, acct, app, host string) (*Record, error) { |
| decoded, err := DecodeRecord(raw) |
| if err != nil { |
| - return nil, raw, etag, fmt.Errorf("provision: record %s/%s/%s: %w: %w", acct, app, host, ErrRecordCorrupt, err) |
| + return nil, fmt.Errorf("provision: record %s/%s/%s: %w: %w", acct, app, host, ErrRecordCorrupt, err) |
| } |
| if decoded.Account != acct || decoded.App != app || decoded.Host != host { |
| - return nil, raw, etag, fmt.Errorf("provision: record %s/%s/%s: %w: record disagrees with its own path", acct, app, host, ErrRecordCorrupt) |
| + return nil, fmt.Errorf("provision: record %s/%s/%s: %w: record disagrees with its own path", acct, app, host, ErrRecordCorrupt) |
| } |
| - return &decoded, raw, etag, nil |
| + return &decoded, nil |
| } |
diff --git a/internal/provision/lifetime_judge.go b/internal/provision/lifetime_judge.go| new file mode 100644 |
| index 000000000..1a9a6c017 |
| --- /dev/null |
| +++ b/internal/provision/lifetime_judge.go |
| @@ -0,0 +1,265 @@ |
| +package provision |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "math" |
| + "os" |
| + "sort" |
| + "sync" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +type Observation struct { |
| + Present bool |
| + Rec *Record |
| + Raw []byte |
| + ETag string |
| + Slot *HostIncarnation |
| + SlotETag string |
| + Verdict LifetimeVerdict |
| + Err error |
| +} |
| + |
| +type LifetimeJudge struct{} |
| + |
| +func NewLifetimeJudge() *LifetimeJudge { return &LifetimeJudge{} } |
| + |
| +// Observe reads the record before the slot so a retired lifetime cannot |
| +// authorize work with bytes read after its replacement. |
| +func (j *LifetimeJudge) Observe(ctx context.Context, s release.Store, acct, app, host string) Observation { |
| + _, raw, etag, err := GetRecordRaw(ctx, s, acct, app, host) |
| + return j.ObserveWith(ctx, s, acct, app, host, raw, etag, err) |
| +} |
| + |
| +// ObserveWith consumes a prefetched record as round one. Each retry replaces |
| +// the whole observation, including the bytes and ETags a caller may act on. |
| +func (j *LifetimeJudge) ObserveWith(ctx context.Context, s release.Store, acct, app, host string, raw []byte, etag string, readErr error) Observation { |
| + var obs Observation |
| + var previousSlotETag string |
| + for round := 0; round < 3; round++ { |
| + if round > 0 { |
| + _, raw, etag, readErr = GetRecordRaw(ctx, s, acct, app, host) |
| + } |
| + obs = Observation{Verdict: LifetimeValid} |
| + switch { |
| + case errors.Is(readErr, os.ErrNotExist): |
| + case readErr != nil && !errors.Is(readErr, ErrRecordCorrupt): |
| + obs.Verdict, obs.Err = LifetimeUnreadable, readErr |
| + return obs |
| + default: |
| + obs.Present, obs.Raw, obs.ETag = true, raw, etag |
| + obs.Rec, obs.Err = decodeRecordAt(raw, acct, app, host) |
| + if obs.Err != nil { |
| + obs.Verdict = LifetimeCorrupt |
| + } |
| + } |
| + if j == nil { |
| + return obs |
| + } |
| + slot, slotETag, err := ReadHostIncarnation(ctx, s, host) |
| + obs.Slot, obs.SlotETag = slot, slotETag |
| + if err != nil { |
| + obs.Verdict, obs.Err = LifetimeUnreadable, err |
| + return obs |
| + } |
| + if obs.Rec != nil && (obs.Rec.Incarnation > math.MaxInt64 || obs.Rec.MoveSeq > math.MaxInt64) { |
| + obs.Verdict, obs.Err = LifetimeUnreadable, fmt.Errorf("%s: %w", host, ErrLifetimeExhausted) |
| + return obs |
| + } |
| + if obs.Verdict == LifetimeCorrupt { |
| + return obs |
| + } |
| + if !obs.Present && slot == nil { |
| + return obs |
| + } |
| + obs.Verdict = LifetimeOf(obs.Rec, obs.ETag, slot) |
| + if obs.Verdict == LifetimeValid { |
| + return obs |
| + } |
| + if round > 0 && slotETag == previousSlotETag { |
| + return obs |
| + } |
| + previousSlotETag = slotETag |
| + } |
| + obs.Verdict, obs.Err = LifetimeUnreadable, fmt.Errorf("%s: %w", host, ErrRecordIndexRetry) |
| + return obs |
| +} |
| + |
| +func observationRecord(obs Observation) (*Record, string, error) { |
| + if obs.Verdict == LifetimeUnreadable || obs.Verdict == LifetimeCorrupt { |
| + return nil, "", obs.Err |
| + } |
| + if !obs.Present || obs.Verdict == LifetimeStale || obs.Rec.Retired { |
| + return nil, "", os.ErrNotExist |
| + } |
| + return obs.Rec, obs.ETag, nil |
| +} |
| + |
| +func ReadRecordJudged(ctx context.Context, s release.Store, j *LifetimeJudge, acct, app, host string) (*Record, string, error) { |
| + if j == nil { |
| + return GetRecordETag(ctx, s, acct, app, host) |
| + } |
| + return observationRecord(j.Observe(ctx, s, acct, app, host)) |
| +} |
| + |
| +func ListRecordsJudged(ctx context.Context, s release.Store, j *LifetimeJudge, acct, app string, max int) (recs []Record, truncated bool, err error) { |
| + recs, truncated, _, err = ListRecordsCompleteJudged(ctx, s, j, acct, app, max) |
| + return recs, truncated, err |
| +} |
| + |
| +func ListRecordsCompleteJudged(ctx context.Context, s release.Store, j *LifetimeJudge, acct, app string, max int) (recs []Record, truncated bool, skipped []string, err error) { |
| + if j == nil { |
| + return ListRecordsComplete(ctx, s, acct, app, max) |
| + } |
| + if max < 0 { |
| + max = 0 |
| + } |
| + prefix := RecordsPrefix(acct, app) |
| + keys, err := s.List(ctx, prefix) |
| + if err != nil { |
| + return nil, false, nil, err |
| + } |
| + hosts := recordHosts(keys, prefix) |
| + sort.Strings(hosts) |
| + if len(hosts) > max { |
| + truncated = true |
| + hosts = hosts[:max] |
| + } |
| + for _, host := range hosts { |
| + obs := j.Observe(ctx, s, acct, app, host) |
| + if obs.Present && obs.Verdict == LifetimeStale { |
| + continue |
| + } |
| + rec, _, err := observationRecord(obs) |
| + if err != nil { |
| + skipped = append(skipped, host) |
| + continue |
| + } |
| + recs = append(recs, *rec) |
| + } |
| + return recs, truncated, skipped, nil |
| +} |
| + |
| +func ListRecordsStrictJudged(ctx context.Context, s release.Store, j *LifetimeJudge, acct, app string, max int) (recs []Record, unreadable map[string]bool, err error) { |
| + if j == nil { |
| + return ListRecordsStrict(ctx, s, acct, app, max) |
| + } |
| + if max < 0 { |
| + max = 0 |
| + } |
| + prefix := RecordsPrefix(acct, app) |
| + keys, err := s.List(ctx, prefix) |
| + if err != nil { |
| + return nil, nil, fmt.Errorf("provision: list records of %s/%s: %w", acct, app, err) |
| + } |
| + hosts := recordHosts(keys, prefix) |
| + if len(hosts) > max { |
| + return nil, nil, ErrRecordsTruncated |
| + } |
| + sort.Strings(hosts) |
| + unreadable = map[string]bool{} |
| + for _, host := range hosts { |
| + obs := j.Observe(ctx, s, acct, app, host) |
| + if obs.Verdict == LifetimeCorrupt { |
| + unreadable[host] = true |
| + continue |
| + } |
| + if !obs.Present || obs.Verdict == LifetimeUnreadable { |
| + _, _, err := observationRecord(obs) |
| + return nil, nil, fmt.Errorf("provision: read record %s: %w", RecordKey(acct, app, host), err) |
| + } |
| + if obs.Verdict == LifetimeStale { |
| + continue |
| + } |
| + recs = append(recs, *obs.Rec) |
| + } |
| + return recs, unreadable, nil |
| +} |
| + |
| +func ListAllRecordsStrictJudged(ctx context.Context, s release.Store, j *LifetimeJudge) (recs []Record, unreadable map[string]bool, err error) { |
| + if j == nil { |
| + return ListAllRecordsStrict(ctx, s) |
| + } |
| + recs, _, unreadable, err = readAllRecordsJudged(ctx, s, j, true) |
| + return recs, unreadable, err |
| +} |
| + |
| +func ListAllRecordsTolerantJudged(ctx context.Context, s release.Store, j *LifetimeJudge) (recs []Record, complete bool, err error) { |
| + if j == nil { |
| + return ListAllRecordsTolerant(ctx, s) |
| + } |
| + recs, complete, _, err = readAllRecordsJudged(ctx, s, j, false) |
| + return recs, complete, err |
| +} |
| + |
| +func readAllRecordsJudged(ctx context.Context, s release.Store, j *LifetimeJudge, strict bool) (recs []Record, complete bool, unreadable map[string]bool, err error) { |
| + complete = true |
| + unreadable = map[string]bool{} |
| + known, readFailed, err := knownAccounts(ctx, s) |
| + if err != nil { |
| + return nil, false, nil, err |
| + } |
| + if readFailed { |
| + if strict { |
| + return nil, false, nil, errors.New("provision: an account manifest could not be read cleanly — refusing to report a partial record set") |
| + } |
| + log.Printf("provision: an account manifest could not be read cleanly: that account's records are skipped") |
| + complete = false |
| + } |
| + keys, partial, err := listAppSubtree(ctx, s, known, "instances/") |
| + if err != nil { |
| + return nil, false, nil, fmt.Errorf("provision: list apps: %w", err) |
| + } |
| + if partial { |
| + if strict { |
| + return nil, false, nil, errors.New("provision: an app's records could not be listed — refusing to report a partial record set") |
| + } |
| + log.Printf("provision: an app's records could not be listed: they are skipped") |
| + complete = false |
| + } |
| + var recordKeys []string |
| + for _, key := range keys { |
| + if _, _, _, ok := ParseRecordKey(key, known); ok { |
| + recordKeys = append(recordKeys, key) |
| + } |
| + } |
| + for _, part := range chunks(recordKeys) { |
| + observations := make(map[string]Observation, len(part)) |
| + var mu sync.Mutex |
| + forEach(part, func(key string) { |
| + acct, app, host, _ := ParseRecordKey(key, known) |
| + obs := j.Observe(ctx, s, acct, app, host) |
| + mu.Lock() |
| + observations[key] = obs |
| + mu.Unlock() |
| + }) |
| + for _, key := range part { |
| + _, _, host, _ := ParseRecordKey(key, known) |
| + obs := observations[key] |
| + if obs.Verdict == LifetimeCorrupt { |
| + log.Printf("provision: skipping record %s: malformed: %v", key, obs.Err) |
| + unreadable[host] = true |
| + complete = false |
| + continue |
| + } |
| + if !obs.Present || obs.Verdict == LifetimeUnreadable { |
| + _, _, err := observationRecord(obs) |
| + if strict { |
| + return nil, false, nil, fmt.Errorf("provision: read record %s: %w", key, err) |
| + } |
| + log.Printf("provision: skipping record %s: unreadable: %v", key, err) |
| + complete = false |
| + continue |
| + } |
| + if obs.Verdict == LifetimeStale { |
| + continue |
| + } |
| + recs = append(recs, *obs.Rec) |
| + } |
| + } |
| + return recs, complete, unreadable, nil |
| +} |
diff --git a/internal/provision/lifetime_judge_test.go b/internal/provision/lifetime_judge_test.go| new file mode 100644 |
| index 000000000..6609b9d52 |
| --- /dev/null |
| +++ b/internal/provision/lifetime_judge_test.go |
| @@ -0,0 +1,760 @@ |
| +package provision |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "math" |
| + "os" |
| + "reflect" |
| + "strings" |
| + "testing" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +const judgeHost = "shop.example.com" |
| + |
| +func judgeRecord() Record { |
| + return Record{Schema: 1, Account: "bab", App: "shop", Host: judgeHost, Channel: "edge", Environment: "production", Incarnation: 1} |
| +} |
| +func judgeSlot(n uint64, state string) HostIncarnation { |
| + return HostIncarnation{Account: "bab", App: "shop", Incarnation: n, State: state} |
| +} |
| +func judgePut(t *testing.T, s release.Store, key string, value any) ([]byte, string) { |
| + t.Helper() |
| + raw, err := json.Marshal(value) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + _, tag, err := s.Get(context.Background(), key) |
| + if err != nil && !errors.Is(err, os.ErrNotExist) { |
| + t.Fatal(err) |
| + } |
| + tag, err = s.Put(context.Background(), key, raw, tag) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return raw, tag |
| +} |
| + |
| +type judgeStore struct { |
| + release.Store |
| + beforeGet func(string) |
| + afterGet func(string) |
| + afterList func(string) |
| + getError func(string) error |
| + puts int |
| +} |
| + |
| +func (s *judgeStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if s.beforeGet != nil { |
| + s.beforeGet(key) |
| + } |
| + if s.getError != nil { |
| + if err := s.getError(key); err != nil { |
| + return nil, "", err |
| + } |
| + } |
| + raw, tag, err := s.Store.Get(ctx, key) |
| + if s.afterGet != nil { |
| + s.afterGet(key) |
| + } |
| + return raw, tag, err |
| +} |
| +func (s *judgeStore) List(ctx context.Context, prefix string) ([]string, error) { |
| + keys, err := s.Store.List(ctx, prefix) |
| + if s.afterList != nil { |
| + s.afterList(prefix) |
| + } |
| + return keys, err |
| +} |
| +func (s *judgeStore) Put(ctx context.Context, key string, body []byte, tag string) (string, error) { |
| + s.puts++ |
| + return s.Store.Put(ctx, key, body, tag) |
| +} |
| +func judgeObserve(s release.Store, j *LifetimeJudge) Observation { |
| + return j.Observe(context.Background(), s, "bab", "shop", judgeHost) |
| +} |
| + |
| +func TestJudgeReadsRecordThenSlot(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + raw, tag := judgePut(t, base, RecordKey("bab", "shop", judgeHost), judgeRecord()) |
| + _, slotTag := judgePut(t, base, HostIncarnationKey(judgeHost), judgeSlot(1, SlotLive)) |
| + var reads []string |
| + s := &judgeStore{Store: base, beforeGet: func(key string) { reads = append(reads, key) }} |
| + obs := judgeObserve(s, NewLifetimeJudge()) |
| + if !reflect.DeepEqual(reads, []string{RecordKey("bab", "shop", judgeHost), HostIncarnationKey(judgeHost)}) { |
| + t.Fatalf("read order %v", reads) |
| + } |
| + if obs.Verdict != LifetimeValid || !obs.Present || obs.Rec == nil || !bytes.Equal(obs.Raw, raw) || obs.ETag != tag || obs.SlotETag != slotTag { |
| + t.Fatalf("observation %+v", obs) |
| + } |
| + reads = nil |
| + obs = NewLifetimeJudge().ObserveWith(context.Background(), s, "bab", "shop", judgeHost, raw, tag, nil) |
| + if !reflect.DeepEqual(reads, []string{HostIncarnationKey(judgeHost)}) || obs.ETag != tag || obs.SlotETag != slotTag { |
| + t.Fatalf("ObserveWith %+v reads %v", obs, reads) |
| + } |
| +} |
| + |
| +func TestJudgeAcrossConcurrentLifetimeChanges(t *testing.T) { |
| + for _, transition := range []string{"retire", "recreate", "migrate"} { |
| + for _, order := range []string{"record-first", "slot-first", "read-straddles", "record-ahead"} { |
| + t.Run(transition+"/"+order, func(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + key := RecordKey("bab", "shop", judgeHost) |
| + old := judgeRecord() |
| + next := old |
| + slot := judgeSlot(1, SlotLive) |
| + switch transition { |
| + case "retire": |
| + next.Retired = true |
| + slot.State = SlotRetired |
| + case "recreate": |
| + next.Incarnation = 2 |
| + slot.Incarnation = 2 |
| + case "migrate": |
| + old.Incarnation = 0 |
| + next = old |
| + next.IncarnationMigrating = "op-1" |
| + slot.Intent = "migration op-1" |
| + } |
| + _, nextTag := judgePut(t, base, key, next) |
| + if transition == "migrate" { |
| + slot.RecordETag = nextTag |
| + } |
| + judgePut(t, base, key, old) |
| + if transition != "migrate" { |
| + judgePut(t, base, HostIncarnationKey(judgeHost), judgeSlot(1, SlotLive)) |
| + } |
| + var wantRaw []byte |
| + var wantTag, wantSlotTag string |
| + installRecord := func() { |
| + wantRaw, wantTag = judgePut(t, base, key, next) |
| + if transition == "migrate" { |
| + slot.RecordETag = wantTag |
| + } |
| + } |
| + installSlot := func() { _, wantSlotTag = judgePut(t, base, HostIncarnationKey(judgeHost), slot) } |
| + reads := 0 |
| + s := &judgeStore{Store: base} |
| + switch order { |
| + case "record-first": |
| + installRecord() |
| + s.beforeGet = func(k string) { |
| + if k == HostIncarnationKey(judgeHost) { |
| + reads++ |
| + if reads == 1 { |
| + installSlot() |
| + } |
| + } |
| + } |
| + case "slot-first": |
| + installSlot() |
| + s.afterGet = func(k string) { |
| + if k == key { |
| + reads++ |
| + if reads == 1 { |
| + installRecord() |
| + } |
| + } |
| + } |
| + case "read-straddles": |
| + s.afterGet = func(k string) { |
| + if k == key { |
| + reads++ |
| + if reads == 1 { |
| + installRecord() |
| + installSlot() |
| + } |
| + } |
| + } |
| + case "record-ahead": |
| + installRecord() |
| + // A migration marker with no slot is already valid. For |
| + // the other transitions, publish the slot after round one. |
| + if transition == "migrate" { |
| + installSlot() |
| + } else { |
| + s.afterGet = func(k string) { |
| + if k == HostIncarnationKey(judgeHost) { |
| + reads++ |
| + if reads == 1 { |
| + installSlot() |
| + } |
| + } |
| + } |
| + } |
| + } |
| + obs := judgeObserve(s, NewLifetimeJudge()) |
| + verdict := LifetimeValid |
| + if transition == "retire" { |
| + verdict = LifetimeStale |
| + } |
| + if obs.Verdict != verdict || !obs.Present || obs.Rec == nil || !bytes.Equal(obs.Raw, wantRaw) || obs.ETag != wantTag || obs.SlotETag != wantSlotTag { |
| + t.Fatalf("observation %+v; want verdict %v, raw %s, tag %s, slot %s", obs, verdict, wantRaw, wantTag, wantSlotTag) |
| + } |
| + }) |
| + } |
| + } |
| + t.Run("slot changes every round", func(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + raw, tag := judgePut(t, base, RecordKey("bab", "shop", judgeHost), judgeRecord()) |
| + count := 0 |
| + lastTag := "" |
| + s := &judgeStore{Store: base, beforeGet: func(k string) { |
| + if k == HostIncarnationKey(judgeHost) { |
| + count++ |
| + _, lastTag = judgePut(t, base, k, judgeSlot(uint64(count+1), SlotRetired)) |
| + } |
| + }} |
| + obs := judgeObserve(s, NewLifetimeJudge()) |
| + if obs.Verdict != LifetimeUnreadable || obs.Err == nil || !strings.Contains(obs.Err.Error(), judgeHost) || count != 3 || !obs.Present || !bytes.Equal(obs.Raw, raw) || obs.ETag != tag || obs.SlotETag != lastTag { |
| + t.Fatalf("exhaustion %+v rounds %d", obs, count) |
| + } |
| + }) |
| +} |
| + |
| +func TestNilJudgeIsClaimsOn(t *testing.T) { |
| + for _, mode := range []string{"valid", "absent", "corrupt", "mismatch", "read-error", "retired"} { |
| + t.Run(mode, func(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + key := RecordKey("bab", "shop", judgeHost) |
| + rec := judgeRecord() |
| + rec.Retired = mode == "retired" |
| + if mode == "mismatch" { |
| + rec.App = "other" |
| + } |
| + if mode != "absent" { |
| + judgePut(t, base, key, rec) |
| + } |
| + if mode == "corrupt" { |
| + putReplace(t, base, key, []byte("broken")) |
| + } |
| + s := &judgeStore{Store: base, getError: func(k string) error { |
| + if k == HostIncarnationKey(judgeHost) { |
| + t.Fatal("nil judge read slot") |
| + } |
| + if mode == "read-error" { |
| + return os.ErrPermission |
| + } |
| + return nil |
| + }} |
| + got, tag, err := ReadRecordJudged(context.Background(), s, nil, "bab", "shop", judgeHost) |
| + want, wtag, werr := GetRecordETag(context.Background(), s, "bab", "shop", judgeHost) |
| + if !reflect.DeepEqual(got, want) || tag != wtag || !sameJudgeError(err, werr) { |
| + t.Fatalf("judged %v %s %v; raw %v %s %v", got, tag, err, want, wtag, werr) |
| + } |
| + obs := judgeObserve(s, nil) |
| + if mode == "absent" && (obs.Present || obs.Verdict != LifetimeValid) { |
| + t.Fatalf("absent %+v", obs) |
| + } |
| + if (mode == "corrupt" || mode == "mismatch") && (!obs.Present || obs.Rec != nil || len(obs.Raw) == 0 || obs.ETag == "" || obs.Verdict != LifetimeCorrupt) { |
| + t.Fatalf("corrupt %+v", obs) |
| + } |
| + if mode == "read-error" && (obs.Present || obs.Verdict != LifetimeUnreadable || !errors.Is(obs.Err, os.ErrPermission)) { |
| + t.Fatalf("error %+v", obs) |
| + } |
| + }) |
| + } |
| + ctx := context.Background() |
| + if WithLifetimes(ctx, nil) != ctx { |
| + t.Fatal("nil WithLifetimes is not identity") |
| + } |
| + j := NewLifetimeJudge() |
| + ctx = WithLifetimes(ctx, j) |
| + if JudgeFrom(ctx) != j || ctx.Value(recordSequenceKey{}) != true { |
| + t.Fatal("lifetime context missing judge or sequence") |
| + } |
| + if JudgeFrom(WithJudge(ctx, nil)) != nil { |
| + t.Fatal("nil judge did not shadow parent") |
| + } |
| +} |
| +func sameJudgeError(a, b error) bool { |
| + if a == nil || b == nil { |
| + return a == b |
| + } |
| + return a.Error() == b.Error() |
| +} |
| + |
| +func TestJudgeUnreadableAndExhausted(t *testing.T) { |
| + for _, mode := range []string{"malformed-slot", "slot-error", "record-error", "record-incarnation", "record-move-seq", "slot-incarnation"} { |
| + t.Run(mode, func(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + rec := judgeRecord() |
| + slot := judgeSlot(1, SlotLive) |
| + if mode == "record-incarnation" { |
| + rec.Incarnation = uint64(math.MaxInt64) + 1 |
| + } |
| + if mode == "record-move-seq" { |
| + rec.MoveSeq = uint64(math.MaxInt64) + 1 |
| + } |
| + if mode == "slot-incarnation" { |
| + slot.Incarnation = uint64(math.MaxInt64) + 1 |
| + } |
| + judgePut(t, base, RecordKey("bab", "shop", judgeHost), rec) |
| + judgePut(t, base, HostIncarnationKey(judgeHost), slot) |
| + if mode == "malformed-slot" { |
| + putReplace(t, base, HostIncarnationKey(judgeHost), []byte("{}")) |
| + } |
| + s := &judgeStore{Store: base, getError: func(k string) error { |
| + if (mode == "slot-error" && k == HostIncarnationKey(judgeHost)) || (mode == "record-error" && k == RecordKey("bab", "shop", judgeHost)) { |
| + return os.ErrPermission |
| + } |
| + return nil |
| + }} |
| + obs := judgeObserve(s, NewLifetimeJudge()) |
| + want := ErrLifetimeExhausted |
| + if mode == "malformed-slot" { |
| + want = ErrSlotMalformed |
| + } |
| + if mode == "slot-error" || mode == "record-error" { |
| + want = os.ErrPermission |
| + } |
| + if obs.Verdict != LifetimeUnreadable || !errors.Is(obs.Err, want) { |
| + t.Fatalf("%+v want %v", obs, want) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestUpdateRecordRefusesAStaleRecordUnderAJudge(t *testing.T) { |
| + for _, kind := range []string{"seed", "region", "move"} { |
| + for _, judged := range []bool{false, true} { |
| + t.Run(kind+"/"+map[bool]string{true: "judged", false: "legacy"}[judged], func(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + raw, tag := judgePut(t, base, RecordKey("bab", "shop", judgeHost), judgeRecord()) |
| + judgePut(t, base, HostIncarnationKey(judgeHost), judgeSlot(2, SlotLive)) |
| + s := &judgeStore{Store: base} |
| + ctx := context.Background() |
| + if judged { |
| + ctx = WithJudge(ctx, NewLifetimeJudge()) |
| + } |
| + called := false |
| + mut := func(r *Record) error { called = true; r.Region = "region-b"; return nil } |
| + var err error |
| + switch kind { |
| + case "seed": |
| + _, err = SetRecordSeed(ctx, s, "bab", "shop", judgeHost, mut) |
| + case "region": |
| + err = SetRecordRegion(ctx, s, "bab", "shop", judgeHost, "region-b") |
| + case "move": |
| + _, err = SetRecordMove(ctx, s, "bab", "shop", judgeHost, mut) |
| + } |
| + if judged { |
| + after, afterTag, _ := base.Get(ctx, RecordKey("bab", "shop", judgeHost)) |
| + if !errors.Is(err, os.ErrNotExist) || called || s.puts != 0 || !bytes.Equal(after, raw) || afterTag != tag { |
| + t.Fatalf("stale mutated: %v called %v puts %d", err, called, s.puts) |
| + } |
| + } else if err != nil || s.puts != 1 { |
| + t.Fatalf("legacy %v puts %d", err, s.puts) |
| + } |
| + }) |
| + } |
| + } |
| +} |
| + |
| +func TestClaimRawUpdateRecordIsJudged(t *testing.T) { |
| + for _, mode := range []string{"stale", "absent", "corrupt", "unreadable", "valid", "etag-conflict", "legacy"} { |
| + t.Run(mode, func(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + rec := judgeRecord() |
| + _, tag := judgePut(t, base, RecordKey("bab", "shop", judgeHost), rec) |
| + slot := judgeSlot(1, SlotLive) |
| + if mode == "stale" || mode == "legacy" { |
| + slot.Incarnation = 2 |
| + } |
| + judgePut(t, base, HostIncarnationKey(judgeHost), slot) |
| + if mode == "absent" { |
| + if err := base.Delete(context.Background(), RecordKey("bab", "shop", judgeHost)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if mode == "corrupt" { |
| + putReplace(t, base, RecordKey("bab", "shop", judgeHost), []byte("broken")) |
| + } |
| + s := &judgeStore{Store: base, getError: func(k string) error { |
| + if mode == "unreadable" && k == HostIncarnationKey(judgeHost) { |
| + return os.ErrPermission |
| + } |
| + return nil |
| + }} |
| + ctx := WithLifetimes(context.Background(), NewLifetimeJudge()) |
| + if mode == "legacy" { |
| + ctx = context.Background() |
| + } |
| + if mode == "etag-conflict" { |
| + tag = "wrong" |
| + } |
| + rec.Region = "region-b" |
| + err := claimRawUpdateRecord(ctx, s, rec, tag) |
| + var want error |
| + switch mode { |
| + case "stale", "absent": |
| + want = os.ErrNotExist |
| + case "corrupt": |
| + want = ErrRecordCorrupt |
| + case "unreadable": |
| + want = os.ErrPermission |
| + case "etag-conflict": |
| + want = release.ErrCASConflict |
| + } |
| + if !errors.Is(err, want) { |
| + t.Fatalf("got %v want %v", err, want) |
| + } |
| + if want != nil && s.puts != 0 { |
| + t.Fatal("refusal wrote record") |
| + } |
| + if mode == "valid" { |
| + got, _ := GetRecord(ctx, base, "bab", "shop", judgeHost) |
| + if got.RecordSeq != 1 { |
| + t.Fatalf("sequence %d", got.RecordSeq) |
| + } |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestListedKeyAbsenceIsNotStaleness(t *testing.T) { |
| + for _, enabled := range []bool{false, true} { |
| + for _, slotExists := range []bool{false, true} { |
| + t.Run(map[bool]string{true: "judged", false: "legacy"}[enabled]+map[bool]string{true: "/slot", false: "/no-slot"}[slotExists], func(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + var j *LifetimeJudge |
| + if enabled { |
| + j = NewLifetimeJudge() |
| + } |
| + key := RecordKey("bab", "shop", judgeHost) |
| + if slotExists { |
| + judgePut(t, base, HostIncarnationKey(judgeHost), judgeSlot(1, SlotRetired)) |
| + } |
| + s := &judgeStore{Store: base, afterList: func(prefix string) { |
| + if strings.HasPrefix(prefix, "apps/") { |
| + if err := base.Delete(context.Background(), key); err != nil && !errors.Is(err, os.ErrNotExist) { |
| + t.Fatal(err) |
| + } |
| + } |
| + }} |
| + for _, kind := range []string{"strict", "complete", "all-strict", "tolerant"} { |
| + judgePut(t, base, key, judgeRecord()) |
| + ctx := context.Background() |
| + switch kind { |
| + case "strict": |
| + _, _, err := ListRecordsStrictJudged(ctx, s, j, "bab", "shop", 10) |
| + if !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("strict %v", err) |
| + } |
| + case "complete": |
| + _, _, skipped, err := ListRecordsCompleteJudged(ctx, s, j, "bab", "shop", 10) |
| + if err != nil || !reflect.DeepEqual(skipped, []string{judgeHost}) { |
| + t.Fatalf("complete %v %v", skipped, err) |
| + } |
| + case "all-strict": |
| + _, _, err := ListAllRecordsStrictJudged(ctx, s, j) |
| + if !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("all strict %v", err) |
| + } |
| + case "tolerant": |
| + _, complete, err := ListAllRecordsTolerantJudged(ctx, s, j) |
| + if err != nil || complete { |
| + t.Fatalf("tolerant %v %v", complete, err) |
| + } |
| + } |
| + } |
| + }) |
| + } |
| + } |
| +} |
| + |
| +func TestJudgedListingsDropStaleKeepValid(t *testing.T) { |
| + for _, bad := range []string{"none", "unreadable", "corrupt"} { |
| + t.Run(bad, func(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + valid := judgeRecord() |
| + judgePut(t, base, RecordKey(valid.Account, valid.App, valid.Host), valid) |
| + judgePut(t, base, HostIncarnationKey(valid.Host), judgeSlot(1, SlotLive)) |
| + stale := valid |
| + stale.Host = "stale.example.com" |
| + judgePut(t, base, RecordKey(stale.Account, stale.App, stale.Host), stale) |
| + judgePut(t, base, HostIncarnationKey(stale.Host), judgeSlot(2, SlotLive)) |
| + broken := valid |
| + broken.Host = "broken.example.com" |
| + if bad != "none" { |
| + judgePut(t, base, RecordKey(broken.Account, broken.App, broken.Host), broken) |
| + if bad == "corrupt" { |
| + putReplace(t, base, RecordKey(broken.Account, broken.App, broken.Host), []byte("broken")) |
| + } |
| + } |
| + s := &judgeStore{Store: base, getError: func(k string) error { |
| + if bad == "unreadable" && k == HostIncarnationKey(broken.Host) { |
| + return os.ErrPermission |
| + } |
| + return nil |
| + }} |
| + ctx := context.Background() |
| + j := NewLifetimeJudge() |
| + check := func(recs []Record) { |
| + t.Helper() |
| + if len(recs) != 1 || recs[0].Host != judgeHost { |
| + t.Fatalf("records %+v", recs) |
| + } |
| + } |
| + recs, truncated, skipped, err := ListRecordsCompleteJudged(ctx, s, j, "bab", "shop", 10) |
| + check(recs) |
| + if err != nil || truncated || len(skipped) != map[bool]int{true: 1, false: 0}[bad != "none"] { |
| + t.Fatalf("complete %v %v %v", truncated, skipped, err) |
| + } |
| + recs, truncated, err = ListRecordsJudged(ctx, s, j, "bab", "shop", 10) |
| + check(recs) |
| + if err != nil || truncated { |
| + t.Fatal(err) |
| + } |
| + for _, all := range []bool{false, true} { |
| + var unreadable map[string]bool |
| + if all { |
| + recs, unreadable, err = ListAllRecordsStrictJudged(ctx, s, j) |
| + } else { |
| + recs, unreadable, err = ListRecordsStrictJudged(ctx, s, j, "bab", "shop", 10) |
| + } |
| + if bad == "unreadable" { |
| + if !errors.Is(err, os.ErrPermission) { |
| + t.Fatalf("strict error %v", err) |
| + } |
| + } else { |
| + check(recs) |
| + if err != nil || unreadable[broken.Host] != (bad == "corrupt") { |
| + t.Fatalf("strict %v %v", unreadable, err) |
| + } |
| + } |
| + } |
| + recs, complete, err := ListAllRecordsTolerantJudged(ctx, s, j) |
| + check(recs) |
| + if err != nil || complete != (bad == "none") { |
| + t.Fatalf("tolerant %v %v", complete, err) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestNilJudgeListingsEqualUnjudged(t *testing.T) { |
| + ctx := context.Background() |
| + for _, mode := range []string{"valid", "corrupt", "mismatch", "retired", "read-error"} { |
| + t.Run(mode, func(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + rec := judgeRecord() |
| + if mode == "retired" { |
| + rec.Retired = true |
| + } |
| + if mode == "mismatch" { |
| + rec.App = "other" |
| + } |
| + judgePut(t, base, RecordKey("bab", "shop", judgeHost), rec) |
| + if mode == "corrupt" { |
| + putReplace(t, base, RecordKey("bab", "shop", judgeHost), []byte("broken")) |
| + } |
| + s := &judgeStore{Store: base, getError: func(key string) error { |
| + if strings.HasPrefix(key, "control/host-incarnations/") { |
| + t.Error("nil listing read a slot") |
| + return os.ErrPermission |
| + } |
| + if mode == "read-error" && key == RecordKey("bab", "shop", judgeHost) { |
| + return os.ErrPermission |
| + } |
| + return nil |
| + }} |
| + for _, cap := range []int{-1, 0, 1, 2} { |
| + got, gt, gs, ge := ListRecordsCompleteJudged(ctx, s, nil, "bab", "shop", cap) |
| + want, wt, ws, we := ListRecordsComplete(ctx, s, "bab", "shop", cap) |
| + if !reflect.DeepEqual(got, want) || gt != wt || !reflect.DeepEqual(gs, ws) || !sameJudgeError(ge, we) { |
| + t.Fatalf("complete differs: %+v %v %v %v / %+v %v %v %v", got, gt, gs, ge, want, wt, ws, we) |
| + } |
| + got, gt, ge = ListRecordsJudged(ctx, s, nil, "bab", "shop", cap) |
| + want, wt, we = ListRecords(ctx, s, "bab", "shop", cap) |
| + if !reflect.DeepEqual(got, want) || gt != wt || !sameJudgeError(ge, we) { |
| + t.Fatal("display differs") |
| + } |
| + got, gu, ge := ListRecordsStrictJudged(ctx, s, nil, "bab", "shop", cap) |
| + want, wu, we := ListRecordsStrict(ctx, s, "bab", "shop", cap) |
| + if !reflect.DeepEqual(got, want) || !reflect.DeepEqual(gu, wu) || !sameJudgeError(ge, we) { |
| + t.Fatal("strict differs") |
| + } |
| + } |
| + got, gu, ge := ListAllRecordsStrictJudged(ctx, s, nil) |
| + want, wu, we := ListAllRecordsStrict(ctx, s) |
| + if !reflect.DeepEqual(got, want) || !reflect.DeepEqual(gu, wu) || !sameJudgeError(ge, we) { |
| + t.Fatal("all strict differs") |
| + } |
| + got, gc, ge := ListAllRecordsTolerantJudged(ctx, s, nil) |
| + want, wc, we := ListAllRecordsTolerant(ctx, s) |
| + if !reflect.DeepEqual(got, want) || gc != wc || !sameJudgeError(ge, we) { |
| + t.Fatal("all tolerant differs") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestJudgeAbsentCorruptAndStableStale(t *testing.T) { |
| + for _, mode := range []string{"absent-no-slot", "absent-live-slot", "absent-retired-slot", "corrupt", "mismatch", "stale", "tombstone"} { |
| + t.Run(mode, func(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + key := RecordKey("bab", "shop", judgeHost) |
| + var raw []byte |
| + var tag, slotTag string |
| + if !strings.HasPrefix(mode, "absent") { |
| + rec := judgeRecord() |
| + if mode == "tombstone" { |
| + rec.Retired = true |
| + } |
| + if mode == "mismatch" { |
| + rec.App = "other" |
| + } |
| + raw, tag = judgePut(t, base, key, rec) |
| + } |
| + if mode == "corrupt" { |
| + raw = []byte("broken") |
| + var err error |
| + tag, err = base.Put(context.Background(), key, raw, tag) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if mode != "absent-no-slot" { |
| + slot := judgeSlot(2, SlotLive) |
| + if mode == "absent-retired-slot" { |
| + slot.State = SlotRetired |
| + } |
| + _, slotTag = judgePut(t, base, HostIncarnationKey(judgeHost), slot) |
| + } |
| + reads := 0 |
| + s := &judgeStore{Store: base, beforeGet: func(k string) { reads++ }} |
| + obs := judgeObserve(s, NewLifetimeJudge()) |
| + verdict := LifetimeStale |
| + rounds := 2 |
| + if mode == "absent-no-slot" { |
| + verdict = LifetimeValid |
| + rounds = 1 |
| + } |
| + if mode == "corrupt" || mode == "mismatch" { |
| + verdict = LifetimeCorrupt |
| + rounds = 1 |
| + } |
| + if obs.Present != !strings.HasPrefix(mode, "absent") || obs.Verdict != verdict || !bytes.Equal(obs.Raw, raw) || obs.ETag != tag || obs.SlotETag != slotTag || reads != 2*rounds { |
| + t.Fatalf("observation %+v reads %d", obs, reads) |
| + } |
| + if (strings.HasPrefix(mode, "absent") || verdict == LifetimeCorrupt) && obs.Rec != nil { |
| + t.Fatal("absent/corrupt observation has record") |
| + } |
| + if verdict == LifetimeCorrupt && !errors.Is(obs.Err, ErrRecordCorrupt) { |
| + t.Fatalf("corrupt error %v", obs.Err) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestJudgeRetryReturnsLatestRecordOnThirdRound(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + key := RecordKey("bab", "shop", judgeHost) |
| + rec := judgeRecord() |
| + judgePut(t, base, key, rec) |
| + var raw []byte |
| + var tag, slotTag string |
| + round := 0 |
| + s := &judgeStore{Store: base, beforeGet: func(k string) { |
| + if k == key { |
| + round++ |
| + if round == 3 { |
| + rec.Incarnation = 3 |
| + raw, tag = judgePut(t, base, key, rec) |
| + } |
| + } |
| + if k == HostIncarnationKey(judgeHost) && round < 3 { |
| + _, slotTag = judgePut(t, base, k, judgeSlot(uint64(round+1), SlotLive)) |
| + } |
| + }} |
| + obs := judgeObserve(s, NewLifetimeJudge()) |
| + if round != 3 || obs.Verdict != LifetimeValid || !obs.Present || !bytes.Equal(obs.Raw, raw) || obs.ETag != tag || obs.SlotETag != slotTag || obs.Rec.Incarnation != 3 { |
| + t.Fatalf("observation %+v rounds %d", obs, round) |
| + } |
| +} |
| + |
| +type judgeConflictStore struct { |
| + release.Store |
| + conflict func() |
| + puts int |
| +} |
| + |
| +func (s *judgeConflictStore) Put(ctx context.Context, key string, raw []byte, etag string) (string, error) { |
| + s.puts++ |
| + if s.puts == 1 { |
| + s.conflict() |
| + return "", release.ErrCASConflict |
| + } |
| + return s.Store.Put(ctx, key, raw, etag) |
| +} |
| +func TestUpdateRecordJudgeRechecksLifetimeAfterCASConflict(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + judgePut(t, base, RecordKey("bab", "shop", judgeHost), judgeRecord()) |
| + judgePut(t, base, HostIncarnationKey(judgeHost), judgeSlot(1, SlotLive)) |
| + s := &judgeConflictStore{Store: base, conflict: func() { judgePut(t, base, HostIncarnationKey(judgeHost), judgeSlot(1, SlotRetired)) }} |
| + calls := 0 |
| + _, err := SetRecordMove(WithLifetimes(context.Background(), NewLifetimeJudge()), s, "bab", "shop", judgeHost, func(r *Record) error { calls++; r.Region = "region-b"; return nil }) |
| + if !errors.Is(err, os.ErrNotExist) || calls != 1 || s.puts != 1 { |
| + t.Fatalf("retry %v calls %d puts %d", err, calls, s.puts) |
| + } |
| +} |
| + |
| +func TestJudgeObserveWithReplacesPrefetchedStaleRead(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + key := RecordKey("bab", "shop", judgeHost) |
| + oldRaw, oldTag := judgePut(t, base, key, judgeRecord()) |
| + next := judgeRecord() |
| + next.Incarnation = 2 |
| + raw, tag := judgePut(t, base, key, next) |
| + _, slotTag := judgePut(t, base, HostIncarnationKey(judgeHost), judgeSlot(2, SlotLive)) |
| + var reads []string |
| + s := &judgeStore{Store: base, beforeGet: func(k string) { reads = append(reads, k) }} |
| + obs := NewLifetimeJudge().ObserveWith(context.Background(), s, "bab", "shop", judgeHost, oldRaw, oldTag, nil) |
| + if !obs.Present || obs.Verdict != LifetimeValid || !bytes.Equal(obs.Raw, raw) || obs.ETag != tag || obs.SlotETag != slotTag || !reflect.DeepEqual(reads, []string{HostIncarnationKey(judgeHost), key, HostIncarnationKey(judgeHost)}) { |
| + t.Fatalf("observation %+v reads %v", obs, reads) |
| + } |
| +} |
| + |
| +func TestJudgeCounterBoundary(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + rec := judgeRecord() |
| + rec.Incarnation = math.MaxInt64 |
| + rec.MoveSeq = math.MaxInt64 |
| + raw, tag := judgePut(t, base, RecordKey("bab", "shop", judgeHost), rec) |
| + _, slotTag := judgePut(t, base, HostIncarnationKey(judgeHost), judgeSlot(math.MaxInt64, SlotLive)) |
| + obs := judgeObserve(base, NewLifetimeJudge()) |
| + if obs.Verdict != LifetimeValid || !obs.Present || !bytes.Equal(obs.Raw, raw) || obs.ETag != tag || obs.SlotETag != slotTag { |
| + t.Fatalf("boundary %+v", obs) |
| + } |
| + judgePut(t, base, HostIncarnationKey(judgeHost), judgeSlot(uint64(math.MaxInt64)+1, SlotLive)) |
| + if _, _, err := ReadHostIncarnation(context.Background(), base, judgeHost); !errors.Is(err, ErrLifetimeExhausted) { |
| + t.Fatalf("slot read %v", err) |
| + } |
| +} |
| + |
| +func TestJudgedListingCapCountsKeys(t *testing.T) { |
| + base := release.NewDirStore(t.TempDir()) |
| + stale := judgeRecord() |
| + stale.Host = "a.example.com" |
| + judgePut(t, base, RecordKey("bab", "shop", stale.Host), stale) |
| + judgePut(t, base, HostIncarnationKey(stale.Host), judgeSlot(2, SlotLive)) |
| + valid := judgeRecord() |
| + judgePut(t, base, RecordKey("bab", "shop", judgeHost), valid) |
| + judgePut(t, base, HostIncarnationKey(judgeHost), judgeSlot(1, SlotLive)) |
| + j := NewLifetimeJudge() |
| + ctx := context.Background() |
| + for _, cap := range []int{-1, 0, 1} { |
| + recs, truncated, skipped, err := ListRecordsCompleteJudged(ctx, base, j, "bab", "shop", cap) |
| + if len(recs) != 0 || !truncated || len(skipped) != 0 || err != nil { |
| + t.Fatalf("cap %d: %v %v %v %v", cap, recs, truncated, skipped, err) |
| + } |
| + if _, _, err := ListRecordsStrictJudged(ctx, base, j, "bab", "shop", cap); !errors.Is(err, ErrRecordsTruncated) { |
| + t.Fatalf("strict cap %d: %v", cap, err) |
| + } |
| + } |
| +} |
diff --git a/internal/provision/provision.go b/internal/provision/provision.go| index 4c618c355..61887cb63 100644 |
| --- a/internal/provision/provision.go |
| +++ b/internal/provision/provision.go |
| @@ -950,21 +950,52 @@ func updateRecord(ctx context.Context, s release.Store, acct, app, host string, |
| return updateRecordSeeding(ctx, s, acct, app, host, mut, false) |
| } |
| +type lifetimeJudgeKey struct{} |
| + |
| +func WithJudge(ctx context.Context, j *LifetimeJudge) context.Context { |
| + return context.WithValue(ctx, lifetimeJudgeKey{}, j) |
| +} |
| + |
| +func JudgeFrom(ctx context.Context) *LifetimeJudge { |
| + j, _ := ctx.Value(lifetimeJudgeKey{}).(*LifetimeJudge) |
| + return j |
| +} |
| + |
| +func WithLifetimes(ctx context.Context, j *LifetimeJudge) context.Context { |
| + if j == nil { |
| + return ctx |
| + } |
| + return WithRecordSequence(WithJudge(ctx, j)) |
| +} |
| + |
| // updateRecordSeeding is updateRecord's scaffold. seedWrite is true only for |
| // the seed transitions: every other caller is refused a change to Seed. |
| func updateRecordSeeding(ctx context.Context, s release.Store, acct, app, host string, mut func(*Record) error, seedWrite bool) (*Record, error) { |
| key := RecordKey(acct, app, host) |
| update := func() (*Record, error) { |
| - raw, etag, err := s.Get(ctx, key) |
| - if err != nil { |
| - return nil, err |
| - } |
| - rec, err := DecodeRecord(raw) |
| - if err != nil { |
| - return nil, fmt.Errorf("provision: record %s/%s/%s: %w", acct, app, host, err) |
| - } |
| - if rec.Account != acct || rec.App != app || rec.Host != host { |
| - return nil, ErrMaintenanceProjection |
| + var raw []byte |
| + var etag string |
| + var rec Record |
| + if j := JudgeFrom(ctx); j != nil { |
| + obs := j.Observe(ctx, s, acct, app, host) |
| + current, _, err := observationRecord(obs) |
| + if err != nil { |
| + return nil, err |
| + } |
| + raw, etag, rec = obs.Raw, obs.ETag, *current |
| + } else { |
| + var err error |
| + raw, etag, err = s.Get(ctx, key) |
| + if err != nil { |
| + return nil, err |
| + } |
| + rec, err = DecodeRecord(raw) |
| + if err != nil { |
| + return nil, fmt.Errorf("provision: record %s/%s/%s: %w", acct, app, host, err) |
| + } |
| + if rec.Account != acct || rec.App != app || rec.Host != host { |
| + return nil, ErrMaintenanceProjection |
| + } |
| } |
| before := rec |
| before.Maintenance = append(MaintenanceProjection(nil), rec.Maintenance...) |
| @@ -2753,16 +2784,32 @@ func claimRawUpdateRecord(ctx context.Context, s release.Store, rec Record, etag |
| return fmt.Errorf("provision: update %s/%s/%s: no etag — an update must be guarded by the read it was decided from", |
| rec.Account, rec.App, rec.Host) |
| } |
| - prior, currentETag, err := s.Get(ctx, RecordKey(rec.Account, rec.App, rec.Host)) |
| - if err != nil { |
| - return err |
| - } |
| - if currentETag != etag { |
| - return release.ErrCASConflict |
| - } |
| - before, err := DecodeRecord(prior) |
| - if err != nil { |
| - return fmt.Errorf("provision: record %s/%s/%s: %w", rec.Account, rec.App, rec.Host, err) |
| + var prior []byte |
| + var currentETag string |
| + var before Record |
| + if j := JudgeFrom(ctx); j != nil { |
| + obs := j.Observe(ctx, s, rec.Account, rec.App, rec.Host) |
| + current, _, err := observationRecord(obs) |
| + if err != nil { |
| + return err |
| + } |
| + prior, currentETag, before = obs.Raw, obs.ETag, *current |
| + if currentETag != etag { |
| + return release.ErrCASConflict |
| + } |
| + } else { |
| + var err error |
| + prior, currentETag, err = s.Get(ctx, RecordKey(rec.Account, rec.App, rec.Host)) |
| + if err != nil { |
| + return err |
| + } |
| + if currentETag != etag { |
| + return release.ErrCASConflict |
| + } |
| + before, err = DecodeRecord(prior) |
| + if err != nil { |
| + return fmt.Errorf("provision: record %s/%s/%s: %w", rec.Account, rec.App, rec.Host, err) |
| + } |
| } |
| if !protectedFieldsEqual(before, rec) { |
| return ErrMaintenanceProjection |
diff --git a/internal/provision/rootrecord.go b/internal/provision/rootrecord.go| index 918a282e3..bd808ce0e 100644 |
| --- a/internal/provision/rootrecord.go |
| +++ b/internal/provision/rootrecord.go |
| @@ -9,7 +9,7 @@ import ( |
| ) |
| type RecordIndex struct { |
| - keys map[string][2]string |
| + keys map[string][][2]string |
| } |
| func IndexRecords(ctx context.Context, s release.Store) (*RecordIndex, error) { |
| @@ -27,17 +27,13 @@ func IndexRecords(ctx context.Context, s release.Store) (*RecordIndex, error) { |
| if partial { |
| return nil, ErrRecordIndexRetry |
| } |
| - out := &RecordIndex{keys: make(map[string][2]string)} |
| + out := &RecordIndex{keys: make(map[string][][2]string)} |
| for _, key := range keys { |
| a, p, h, ok := ParseRecordKey(key, known) |
| if !ok { |
| continue |
| } |
| - if _, duplicate := out.keys[h]; duplicate { |
| - out.keys[h] = [2]string{} |
| - } else { |
| - out.keys[h] = [2]string{a, p} |
| - } |
| + out.keys[h] = append(out.keys[h], [2]string{a, p}) |
| } |
| return out, nil |
| } |
| @@ -54,14 +50,31 @@ func (i *RecordIndex) Find(ctx context.Context, s release.Store, host string) (* |
| if !ok { |
| return nil, "", os.ErrNotExist |
| } |
| - account, app := names[0], names[1] |
| - if account == "" { |
| + j := JudgeFrom(ctx) |
| + if j == nil && len(names) > 1 { |
| return nil, "", os.ErrPermission |
| } |
| - rec, etag, err := GetRecordETag(ctx, s, account, app, host) |
| - if err != nil { |
| - return nil, "", err |
| + var rec *Record |
| + var etag string |
| + valid := 0 |
| + for _, candidate := range names { |
| + obs := j.Observe(ctx, s, candidate[0], candidate[1], host) |
| + if obs.Verdict == LifetimeCorrupt || obs.Verdict == LifetimeUnreadable { |
| + return nil, "", obs.Err |
| + } |
| + if !obs.Present || obs.Verdict == LifetimeStale || obs.Rec.Retired { |
| + continue |
| + } |
| + valid++ |
| + rec, etag = obs.Rec, obs.ETag |
| + } |
| + if valid == 0 { |
| + return nil, "", os.ErrNotExist |
| + } |
| + if valid > 1 { |
| + return nil, "", os.ErrPermission |
| } |
| + account, app := rec.Account, rec.App |
| if etag == "" || rec.Schema != 1 || !release.ValidAccount(account) || !release.ValidApp(app) { |
| return nil, "", os.ErrPermission |
| } |
diff --git a/internal/provision/rootrecord_test.go b/internal/provision/rootrecord_test.go| index cc28461fe..bec092534 100644 |
| --- a/internal/provision/rootrecord_test.go |
| +++ b/internal/provision/rootrecord_test.go |
| @@ -3,6 +3,8 @@ package provision |
| import ( |
| "context" |
| "encoding/json" |
| + "errors" |
| + "os" |
| "testing" |
| "github.com/carlosframework/platform/internal/release" |
| @@ -50,3 +52,77 @@ func TestPlan14ProofRecordLookup(t *testing.T) { |
| t.Fatal("key/body mismatch bound") |
| } |
| } |
| + |
| +func TestIndexFindJudgesCandidates(t *testing.T) { |
| + for _, mode := range []string{"successor", "two-legacy", "corrupt", "mismatch", "unreadable", "all-stale", "nil-ambiguous"} { |
| + t.Run(mode, func(t *testing.T) { |
| + ctx := WithJudge(context.Background(), NewLifetimeJudge()) |
| + if mode == "nil-ambiguous" { |
| + ctx = context.Background() |
| + } |
| + base := release.NewDirStore(t.TempDir()) |
| + old := judgeRecord() |
| + next := old |
| + next.App = "successor" |
| + if mode == "two-legacy" { |
| + old.Incarnation = 0 |
| + next.Incarnation = 0 |
| + } |
| + judgePut(t, base, RecordKey(old.Account, old.App, old.Host), old) |
| + _, tag := judgePut(t, base, RecordKey(next.Account, next.App, next.Host), next) |
| + slot := judgeSlot(1, SlotLive) |
| + slot.App = next.App |
| + if mode == "all-stale" { |
| + slot.State = SlotRetired |
| + } |
| + if mode != "two-legacy" { |
| + judgePut(t, base, HostIncarnationKey(judgeHost), slot) |
| + } |
| + if mode == "corrupt" { |
| + putReplace(t, base, RecordKey(old.Account, old.App, old.Host), []byte("broken")) |
| + } |
| + if mode == "mismatch" { |
| + old.Host = "wrong.example.com" |
| + judgePut(t, base, RecordKey("bab", "shop", judgeHost), old) |
| + } |
| + s := &judgeStore{Store: base, getError: func(key string) error { |
| + if mode == "unreadable" && key == RecordKey("bab", "shop", judgeHost) { |
| + return os.ErrPermission |
| + } |
| + return nil |
| + }} |
| + index, err := IndexRecords(ctx, s) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, find := range []func(context.Context, release.Store, string) (*Record, string, error){index.Find, FindRecordByHost} { |
| + rec, gotTag, err := find(ctx, s, judgeHost) |
| + switch mode { |
| + case "successor": |
| + if err != nil || rec.App != next.App || gotTag != tag { |
| + t.Fatalf("successor %+v %s %v", rec, gotTag, err) |
| + } |
| + case "two-legacy", "nil-ambiguous", "unreadable": |
| + if !errors.Is(err, os.ErrPermission) { |
| + t.Fatalf("got %v", err) |
| + } |
| + case "corrupt", "mismatch": |
| + if !errors.Is(err, ErrRecordCorrupt) { |
| + t.Fatalf("got %v", err) |
| + } |
| + case "all-stale": |
| + if !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("got %v", err) |
| + } |
| + } |
| + } |
| + if mode == "successor" { |
| + slot.State = SlotRetired |
| + judgePut(t, base, HostIncarnationKey(judgeHost), slot) |
| + if _, _, err := index.Find(ctx, s, judgeHost); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("index cached verdict: %v", err) |
| + } |
| + } |
| + }) |
| + } |
| +} |
diff --git a/internal/provision/self_serve_moves1_census_test.go b/internal/provision/self_serve_moves1_census_test.go| index 602cd25f5..6f87d8718 100644 |
| --- a/internal/provision/self_serve_moves1_census_test.go |
| +++ b/internal/provision/self_serve_moves1_census_test.go |
| @@ -10,17 +10,42 @@ package provision |
| // - provision.go:GetRecordETag is removed: it no longer builds a record key. |
| // It reads through GetRecordRaw and hides a tombstone, and it writes nothing. |
| // - lifetime.go:GetRecordRaw is a key-site (RecordKey): the raw reader that |
| -// sees a tombstone and a corrupt body. A read only. |
| +// sees a tombstone and a corrupt body. A read only; Task 6a shares its |
| +// decoding and path checks with ObserveWith through decodeRecordAt. |
| // - lifetime.go:WriteHostIncarnation is nonrecord: it Puts only |
| // control/host-incarnations/<host>.json, conditionally (create-only or |
| // If-Match). The key is not an instance or sidecar record key, and the |
| // unit names no record-key constructor or record literal. |
| +// |
| +// Task 6a, reviewed dispositions (all remain PROPOSED): |
| +// - updateRecordSeeding: record-writer; judge each read, including a CAS |
| +// retry, before invoking mut; write only through WriteRecord. |
| +// - claimRawUpdateRecord: record-writer; judge the re-read before the |
| +// caller's ETag check and preserve the existing WriteRecord CAS. |
| +// - IndexRecords: key-site; retain all candidate owner keys for each host; |
| +// no body or verdict is cached, and this function writes nothing. |
| +// - ListRecordsCompleteJudged: key-site; list the app prefix, cap by keys, |
| +// and omit only present stale bodies; missing reads remain skipped. |
| +// - ListRecordsStrictJudged: key-site; list the app prefix, fail on reads |
| +// or truncation, name corrupt hosts, and omit present stale bodies. |
| +// - readAllRecordsJudged: key-site; discover known-account record keys and |
| +// judge them with bounded parallel reads; strict errors and tolerant |
| +// incompleteness preserve gaps. None of the judged listings writes. |
| func plan14SelfServeMoves1WriteCensusInventory() []censusEntry { |
| removed := []struct{ file, unit string }{ |
| {"internal/provision/provision.go", "GetRecordETag"}, |
| + {"internal/provision/provision.go", "updateRecordSeeding"}, |
| + {"internal/provision/provision.go", "claimRawUpdateRecord"}, |
| + {"internal/provision/rootrecord.go", "IndexRecords"}, |
| } |
| added := []censusEntry{ |
| - {"internal/provision/lifetime.go", "GetRecordRaw", "key-site", "key:RecordKey", "e4c422472e601a14496af407c4f9d0535a48809964d6dfbf6b06a7c2e125f077"}, |
| + {"internal/provision/provision.go", "updateRecordSeeding", "record-writer", "key:RecordKey,key:WriteRecord", "20e3b0498fb8d410c3d10bc7c788b0eec137cf58aa4dfa2afb42da837ebd71ea"}, |
| + {"internal/provision/provision.go", "claimRawUpdateRecord", "record-writer", "key:RecordKey,key:WriteRecord,key:claimRawUpdateRecord", "782c1020ceea3661d972715f2ce2500119ebd56eb8e15f89c7a28d79d83758ba"}, |
| + {"internal/provision/rootrecord.go", "IndexRecords", "key-site", "key:ParseRecordKey,lit:instances", "e44ac431865bf38baa4105f66c9cb23014cc3c7ffcd41d5385244cc76fc4c0b9"}, |
| + {"internal/provision/lifetime_judge.go", "ListRecordsCompleteJudged", "key-site", "key:RecordsPrefix", "d851c7d8d04be82fbda88e257bc02825aff67c41b8a1f029f1a8c49f6fe4516c"}, |
| + {"internal/provision/lifetime_judge.go", "ListRecordsStrictJudged", "key-site", "key:RecordKey,key:RecordsPrefix", "95eb5de787af1cabccdc2c3ae56da5226ee876f203d253a2e8335322b20a5d5b"}, |
| + {"internal/provision/lifetime_judge.go", "readAllRecordsJudged", "key-site", "key:ParseRecordKey,lit:instances", "b7be8d74718cfaa96220d2a070b619266167086e69881baa621c4af4c7413aa7"}, |
| + {"internal/provision/lifetime.go", "GetRecordRaw", "key-site", "key:RecordKey", "1e893cadc5dedac4687a6877457254c8e07c328a91b2c5923ef87fd3c41ff1fc"}, |
| {"internal/provision/lifetime.go", "WriteHostIncarnation", "nonrecord", "sel:Put", "c5f2719a53447c0477706600081a9846444071f39b85759a463d1ee757dc760e"}, |
| } |
| entries := plan14ClaimRosterBindingWriteCensusInventory() |
diff --git a/internal/provision/testdata/plan14-root-reads-self-serve-moves-1.tsv b/internal/provision/testdata/plan14-root-reads-self-serve-moves-1.tsv| index 18510bd32..dbe8a80ea 100644 |
| --- a/internal/provision/testdata/plan14-root-reads-self-serve-moves-1.tsv |
| +++ b/internal/provision/testdata/plan14-root-reads-self-serve-moves-1.tsv |
| @@ -1,5 +1,5 @@ |
| # Census overlay self-serve-moves-1, marked rev3=PROPOSED-self-serve-moves-1 (not an accepted ruling; docs/superpowers/notes/census-proposal-self-serve-moves-1.md). Task 1 adds the lifetime fields to provision.Record (incarnation, incarnation_migrating, move_seq, retired) and internal/provision/lifetime.go (the host incarnation slot, the lifetime verdict, the tombstone body, GetRecordRaw); GetRecordETag now hides a tombstone. Envelope only: no new root read, class or task, and the registry schema (@fields) is unchanged because Record is not a registry struct. internal/provision is re-pinned over the new code. The seal stands. |
| -@coverage internal/provision a7dc8a5a3c06f69678b8e337a5cb72097070d813d8e69f2e1b4742e07ef3b5c8 |
| +@coverage internal/provision 7ab4023fb302f730f830f24ecb5a5095bfbc8b26edf6482b9acc69d6e4596338 |
| # Task 2 and fix round 1: rev3=PROPOSED-self-serve-moves-1 throughout; all dispositions below await the human census ruling. Console SQL and randomness run as carlos-console under console.service; no root decision is added. Existing legacy rows are re-frozen at this boundary, and the two constructors lose their own read sites to insertAppRow. |
| # The H3 removal syntax requires the census-ruling- prefix; PROPOSED in those tokens is deliberate and does not claim approval. |
| @legacy 296610cb0201ccb67b21152a50191b8b819f41e1d39fd43ca8b8d35cd75a2601 internal/console/lifetimes_store.go Store.BeginAllocation legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=console.service user carlos-console, J4 root-input: inserts one pending allocation in the console SQLite store, carrying the account, app and app lifetime; no bucket or root action file:Exec@132 7c7cb202595de014e2a549ebb001ae2c0afd800fc7959a6b5e145f5c1bc9c8ef |
| @@ -46,3 +46,6 @@ internal/console/store_gate.go claimSQLCensus legacy accepted-legacy accepted-le |
| @removed internal/console/tempstore.go Store.CreateTemporaryAppInRegion H3 census-ruling-PROPOSED-self-serve-moves-1;rev3=PROPOSED-self-serve-moves-1;basis=constructor remains declared but has no direct read; app insertion and lifetime minting moved to insertAppRow, classified above; proposed removal only |
| # Task 2 envelopes, rev3=PROPOSED-self-serve-moves-1: internal/console covers the lifetime store, app constructors, lifetime projection and table-scoped gate labels; internal/provision (above) adds only ErrLifetimeExhausted beyond Task 1. Registry fields are unchanged. |
| @coverage internal/console fdb47efcb501addb7b7b0bfc756247c64494153e6cbc20d50e5873e4e011ac87 |
| +# Task 6a: rev3=PROPOSED-self-serve-moves-1; reviewed provision envelope now includes bounded record-then-slot observations, judged read/list/update gates and signed-counter exhaustion at slot read. No removal-loop, host or console change; registry fields and the legacy seal stand. |
| +internal/provision/rootrecord.go IndexRecords authority B3,C rev3=PROPOSED-self-serve-moves-1;basis=retain every discovered candidate owner key per host for fresh lifetime judgement at Find; partial discovery still retries and no verdict or body is cached e44ac431865bf38baa4105f66c9cb23014cc3c7ffcd41d5385244cc76fc4c0b9 |
| +internal/provision/rootrecord.go RecordIndex.Find authority B3,C rev3=PROPOSED-self-serve-moves-1;basis=observe each candidate through the context judge, refuse corruption and read failures, require one valid owner and validate its schema and shape; nil-judge duplicates retain permission refusal 760eed59cbd6a8ddab1098ce3a5f7a4272ecda81c17cc079dc135aba3a85a8f8 |