amadan
fix-reconcile-parallel-reads
Squashed from fix-reconcile-parallel-reads into main.
3 files changed,
+683
−215
internal/provision/parallelreads_test.go+320 −0internal/provision/reconcile.go+272 −130internal/provision/sidecar_reconcile.go+91 −85
CI passed — run details
diff --git a/internal/provision/parallelreads_test.go b/internal/provision/parallelreads_test.go| new file mode 100644 |
| index 000000000..0c01a24a8 |
| --- /dev/null |
| +++ b/internal/provision/parallelreads_test.go |
| @@ -0,0 +1,320 @@ |
| +package provision |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// overlapStore holds each record Get open briefly and records the most Gets |
| +// it saw in flight at once. |
| +type overlapStore struct { |
| + release.Store |
| + mu sync.Mutex |
| + inFlight int |
| + maxFlight int |
| + reads int |
| + failKey string |
| +} |
| + |
| +func (s *overlapStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if strings.Contains(key, "/instances/") { |
| + s.mu.Lock() |
| + s.inFlight++ |
| + s.reads++ |
| + if s.inFlight > s.maxFlight { |
| + s.maxFlight = s.inFlight |
| + } |
| + s.mu.Unlock() |
| + time.Sleep(20 * time.Millisecond) |
| + s.mu.Lock() |
| + s.inFlight-- |
| + s.mu.Unlock() |
| + if key == s.failKey { |
| + return nil, "", errors.New("read refused") |
| + } |
| + } |
| + return s.Store.Get(ctx, key) |
| +} |
| + |
| +func seedRecords(t *testing.T, n int) (release.Store, string) { |
| + t.Helper() |
| + inner, sqid := setup(t) |
| + for i := 0; i < n; i++ { |
| + rec := acmeRecord(sqid) |
| + slug := "r" + string(rune('a'+i)) |
| + rec.Host, rec.Slug = slug+".go.tito.io", slug |
| + if err := CreateRecord(ctx, inner, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + return inner, sqid |
| +} |
| + |
| +// The record readers read records several at a time. One at a time, a few |
| +// hundred records cost the fast loop most of a 27-second pass on every |
| +// serial bump (#231). |
| +func TestRecordReadersReadRecordsConcurrently(t *testing.T) { |
| + inner, _ := seedRecords(t, 12) |
| + cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} |
| + for _, c := range []struct { |
| + name string |
| + run func(release.Store) error |
| + }{ |
| + {"Reconcile", func(s release.Store) error { return Reconcile(ctx, s, regWith(t), cfg) }}, |
| + {"ListAllRecords", func(s release.Store) error { _, err := ListAllRecords(ctx, s); return err }}, |
| + } { |
| + s := &overlapStore{Store: inner} |
| + if err := c.run(s); err != nil { |
| + t.Fatalf("%s: %v", c.name, err) |
| + } |
| + if s.maxFlight < 2 || s.maxFlight > 8 { |
| + t.Fatalf("%s read up to %d records at a time, want several and never more than 8", c.name, s.maxFlight) |
| + } |
| + if s.reads != 12 { |
| + t.Fatalf("%s made %d record reads for 12 records, want one each", c.name, s.reads) |
| + } |
| + } |
| +} |
| + |
| +// A read that fails is still that record's failure, judged where it always |
| +// was: reconcile keeps the routes it could not read, and mints the rest. |
| +func TestReconcileStillJudgesAFailedReadPerRecord(t *testing.T) { |
| + inner, sqid := seedRecords(t, 4) |
| + cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} |
| + reg := regWith(t) |
| + if err := Reconcile(ctx, inner, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + s := &overlapStore{Store: inner, failKey: RecordKey(sqid, "tito", "rb.go.tito.io")} |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + // A record created since the first pass must still be minted in a pass |
| + // where another record's read fails: the failure is rb's alone. |
| + fresh := acmeRecord(sqid) |
| + fresh.Host, fresh.Slug = "re.go.tito.io", "re" |
| + if err := CreateRecord(ctx, inner, fresh); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, h := range []string{"ra.go.tito.io", "rb.go.tito.io", "rc.go.tito.io", "rd.go.tito.io", "re.go.tito.io"} { |
| + if rt, err := reg.Get(h); err != nil || rt == nil { |
| + t.Fatalf("%s missing after a pass where only rb's read failed: %+v, %v", h, rt, err) |
| + } |
| + } |
| + _, err := ListAllRecords(ctx, s) |
| + if err == nil || !strings.Contains(err.Error(), "rb.go.tito.io") { |
| + t.Fatalf("ListAllRecords with rb unreadable = %v, want an error naming rb", err) |
| + } |
| +} |
| + |
| +// A loop that stops at its first bad read has read at most one chunk ahead, |
| +// and a large record set is never prefetched whole. |
| +func TestListAllRecordsStopsWithinAChunkOfABadRead(t *testing.T) { |
| + inner, sqid := setup(t) |
| + const total = 200 // more than one chunk, fixed so the test does not move with readChunk |
| + for i := 0; i < total; i++ { |
| + rec := acmeRecord(sqid) |
| + rec.Host, rec.Slug = fmt.Sprintf("r%04d.go.tito.io", i), fmt.Sprintf("r%04d", i) |
| + if err := CreateRecord(ctx, inner, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + s := &countingGets{Store: inner, failKey: RecordKey(sqid, "tito", "r0000.go.tito.io")} |
| + if _, err := ListAllRecords(ctx, s); err == nil { |
| + t.Fatal("ListAllRecords read past an unreadable first record without an error") |
| + } |
| + if s.n > readChunk || s.n >= total { |
| + t.Fatalf("read %d of %d records before stopping at the first, want at most one chunk (%d)", s.n, total, readChunk) |
| + } |
| +} |
| + |
| +// A cancelled context reads nothing, and every key reports why. |
| +func TestGetEachOnACancelledContextReadsNothing(t *testing.T) { |
| + inner, sqid := setup(t) |
| + s := &countingGets{Store: inner} |
| + cctx, cancel := context.WithCancel(ctx) |
| + cancel() |
| + key := RecordKey(sqid, "tito", "acme.go.tito.io") |
| + got := getEach(cctx, s, []string{key}) |
| + if s.n != 0 { |
| + t.Fatalf("made %d reads on a cancelled context", s.n) |
| + } |
| + if !errors.Is(got[key].err, context.Canceled) { |
| + t.Fatalf("key reported %v, want context.Canceled", got[key].err) |
| + } |
| +} |
| + |
| +type countingGets struct { |
| + release.Store |
| + mu sync.Mutex |
| + n int |
| + failKey string |
| +} |
| + |
| +func (s *countingGets) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if strings.Contains(key, "/instances/") { |
| + s.mu.Lock() |
| + s.n++ |
| + s.mu.Unlock() |
| + if key == s.failKey { |
| + return nil, "", errors.New("read refused") |
| + } |
| + } |
| + return s.Store.Get(ctx, key) |
| +} |
| + |
| +// appOverlapStore holds each app-level read (an app's instances manifest and |
| +// its trash marker) open briefly and records the most in flight at once. |
| +type appOverlapStore struct { |
| + release.Store |
| + mu sync.Mutex |
| + inFlight int |
| + maxFlight int |
| + reads map[string]int |
| +} |
| + |
| +func (s *appOverlapStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if strings.HasSuffix(key, "/instances.json") || strings.HasSuffix(key, "/deleted.json") { |
| + s.mu.Lock() |
| + s.inFlight++ |
| + if s.inFlight > s.maxFlight { |
| + s.maxFlight = s.inFlight |
| + } |
| + if s.reads == nil { |
| + s.reads = map[string]int{} |
| + } |
| + s.reads[key]++ |
| + s.mu.Unlock() |
| + time.Sleep(20 * time.Millisecond) |
| + s.mu.Lock() |
| + s.inFlight-- |
| + s.mu.Unlock() |
| + } |
| + return s.Store.Get(ctx, key) |
| +} |
| + |
| +// Reconcile reads each app's manifest and trash marker several apps at a |
| +// time, and each one once. |
| +func TestReconcileReadsAppStateConcurrentlyAndOnce(t *testing.T) { |
| + inner, sqid := setup(t) |
| + want := map[string]bool{} |
| + for i := 0; i < 10; i++ { |
| + app := "app" + string(rune('a'+i)) |
| + want[release.AppPrefix(sqid, app)+"instances.json"] = true |
| + want[release.AppTombstoneKey(sqid, app)] = true |
| + if err := PutManifest(ctx, inner, sqid, app, Manifest{Schema: 1, App: app, Domain: app + ".example", Pool: "tito"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + rec := acmeRecord(sqid) |
| + rec.App, rec.Host, rec.Slug = app, "x."+app+".example", "x" |
| + if err := CreateRecord(ctx, inner, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + s := &appOverlapStore{Store: inner} |
| + cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} |
| + if err := Reconcile(ctx, s, regWith(t), cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if s.maxFlight < 2 || s.maxFlight > 8 { |
| + t.Fatalf("app state read up to %d at a time, want several and never more than 8", s.maxFlight) |
| + } |
| + for key := range want { |
| + if s.reads[key] != 1 { |
| + t.Fatalf("%s read %d times in one pass, want once", key, s.reads[key]) |
| + } |
| + } |
| +} |
| + |
| +// An app in the trash is never minted for, so its manifest is not read: the |
| +// order the record loop always asked in, trash first. |
| +func TestReconcileDoesNotReadATrashedAppsManifest(t *testing.T) { |
| + inner, sqid := setup(t) |
| + tomb := fmt.Sprintf(`{"schema":1,"account":%q,"app":"tito","deleted_at":"2026-10-05T00:00:00Z","actor":"test"}`, sqid) |
| + if _, err := inner.Put(ctx, release.AppTombstoneKey(sqid, "tito"), []byte(tomb), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got, err := release.ReadAppTombstone(ctx, inner, sqid, "tito"); err != nil || got == nil { |
| + t.Fatalf("fixture: the trash marker does not read as one: %+v, %v", got, err) |
| + } |
| + if err := CreateRecord(ctx, inner, acmeRecord(sqid)); err != nil { |
| + t.Fatal(err) |
| + } |
| + s := &appOverlapStore{Store: inner} |
| + cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} |
| + if err := Reconcile(ctx, s, regWith(t), cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if n := s.reads[release.AppPrefix(sqid, "tito")+"instances.json"]; n != 0 { |
| + t.Fatalf("a trashed app's manifest was read %d times", n) |
| + } |
| +} |
| + |
| +// Every chunk is walked: 130 records, three chunks, each record minted and |
| +// listed, each read once. |
| +func TestRecordReadersWalkEveryChunk(t *testing.T) { |
| + inner, sqid := setup(t) |
| + const total = 130 |
| + hosts := map[string]bool{} |
| + for i := 0; i < total; i++ { |
| + rec := acmeRecord(sqid) |
| + rec.Host, rec.Slug = fmt.Sprintf("r%04d.go.tito.io", i), fmt.Sprintf("r%04d", i) |
| + hosts[rec.Host] = true |
| + if err := CreateRecord(ctx, inner, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + s := &perKeyGets{Store: inner} |
| + reg := regWith(t) |
| + cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + for h := range hosts { |
| + if rt, err := reg.Get(h); err != nil || rt == nil { |
| + t.Fatalf("%s not minted: %+v, %v", h, rt, err) |
| + } |
| + if n := s.count(RecordKey(sqid, "tito", h)); n != 1 { |
| + t.Fatalf("%s read %d times by Reconcile, want once", h, n) |
| + } |
| + } |
| + recs, err := ListAllRecords(ctx, s) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(recs) != total { |
| + t.Fatalf("ListAllRecords = %d records, want %d", len(recs), total) |
| + } |
| +} |
| + |
| +type perKeyGets struct { |
| + release.Store |
| + mu sync.Mutex |
| + n map[string]int |
| +} |
| + |
| +func (s *perKeyGets) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + s.mu.Lock() |
| + if s.n == nil { |
| + s.n = map[string]int{} |
| + } |
| + s.n[key]++ |
| + s.mu.Unlock() |
| + return s.Store.Get(ctx, key) |
| +} |
| + |
| +func (s *perKeyGets) count(key string) int { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + return s.n[key] |
| +} |
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go| index 4261942b6..88617b7c1 100644 |
| --- a/internal/provision/reconcile.go |
| +++ b/internal/provision/reconcile.go |
| @@ -724,95 +724,101 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| // back — every other record still converges/removes normally the same |
| // pass. |
| protect := map[string]bool{} |
| + var recordKeys []string |
| for _, key := range keys { |
| - acct, app, host, ok := ParseRecordKey(key, known) |
| - if !ok { |
| - continue |
| - } |
| - raw, _, err := s.Get(ctx, key) |
| - if err != nil { |
| - log.Printf("provision: skipping record %s: read failed: %v", key, err) |
| - // The record's absence from recs below must not read as "the |
| - // bucket retired it" — a transient Get failure disables removal |
| - // for the whole pass, same as an unreadable account manifest. |
| - readFailed = true |
| - continue |
| - } |
| - rec, err := DecodeRecord(raw) |
| - if err != nil { |
| - log.Printf("provision: skipping record %s: malformed: %v", key, err) |
| - protect[host] = true |
| - continue |
| - } |
| - if rec.Host != host || rec.Account != acct || rec.App != app { |
| - log.Printf("provision: skipping record %s: disagrees with its own path (host=%q account=%q app=%q)", |
| - key, rec.Host, rec.Account, rec.App) |
| - protect[host] = true |
| - continue |
| - } |
| - // The record IS the authorization, and this pass turns its host into |
| - // a socket path and a database path (DefaultSocket/DefaultDBPath) |
| - // and a registry primary key. The console's create door enforces |
| - // this same shape — but a door is not a guarantee about what is |
| - // already in the bucket: anything with bucket write access, and any |
| - // record written by a different (older, or future) writer, arrives |
| - // here unexamined. Both doors hold the invariant, neither trusts |
| - // the other. Protected like a corrupted body, and for the same |
| - // reason: a record that cannot be used is not a record that was |
| - // retired. |
| - if !ValidHostShape(rec.Host) { |
| - log.Printf("provision: skipping record %s: host %q is not a hostname", key, rec.Host) |
| - protect[host] = true |
| - continue |
| - } |
| - // ValidateShape is the console create door's policy check — but a |
| - // door is not a guarantee about what is already in the bucket (the |
| - // same reasoning as the host-shape check just above). This pass |
| - // materializes whatever it finds, so a record whose Kind/Addr/Unit/ |
| - // Environment combination is malformed must be caught here too, or |
| - // a bad value becomes a live route. Protected like a corrupted |
| - // body: the record is there, it just cannot be used. |
| - if err := ValidateShape(rec); err != nil { |
| - log.Printf("provision: skipping record %s: invalid shape: %v", key, err) |
| - protect[host] = true |
| - continue |
| - } |
| - // ValidateShape deliberately does not inspect Channel (it validates |
| - // shape fields, and Channel is not one), but a channel-less record |
| - // must not become — or converge — a route pinned to no channel: |
| - // that is issue #130's failure mode arriving through a different |
| - // door than the one Task 3 closed. Protected like a corrupted body: |
| - // the record is there, it just cannot be used. |
| - if rec.Channel == "" { |
| - log.Printf("provision: skipping record %s: empty channel", key) |
| - protect[host] = true |
| - continue |
| - } |
| - // Scope. For a home-boxed record this is the region/pool test it |
| - // always was; for a static or public-addr one the region half widens |
| - // to every platform box in the deployment (everywhereClass, above). |
| - // A replica record narrows it: it belongs ONLY to the box whose |
| - // placement is the record's own Region — the placement the create |
| - // door validated as armed — so a replica record minted on the wrong |
| - // region would mint a route with no mirror behind it and no lease |
| - // speaking to it, a process holding live order data nobody |
| - // authorized into that region (the residence rule, #162). Every |
| - // other box skips it exactly like any other out-of-scope record; |
| - // the replica's own box sees Region == cfg.Region through the |
| - // ordinary test above. |
| - if !cfg.inScope(rec.Region, rec.Pool, everywhereClass(rec)) { |
| - continue // another agent's jurisdiction, not this pass's business |
| + if _, _, _, ok := ParseRecordKey(key, known); ok { |
| + recordKeys = append(recordKeys, key) |
| } |
| - // Two accounts writing the same host is a collision this pass |
| - // cannot arbitrate silently: whichever record this listing reaches |
| - // FIRST wins the host, and the loser is skipped loudly rather than |
| - // overwriting the winner in the map with no trace of the conflict. |
| - if prev, dup := recs[rec.Host]; dup { |
| - log.Printf("provision: duplicate host %s: keeping %s, skipping %s", rec.Host, |
| - RecordKey(prev.Account, prev.App, prev.Host), RecordKey(rec.Account, rec.App, rec.Host)) |
| - continue |
| + } |
| + for _, part := range chunks(recordKeys) { |
| + bodies := getEach(ctx, s, part) |
| + for _, key := range part { |
| + acct, app, host, _ := ParseRecordKey(key, known) |
| + raw, err := bodies[key].body, bodies[key].err |
| + if err != nil { |
| + log.Printf("provision: skipping record %s: read failed: %v", key, err) |
| + // The record's absence from recs below must not read as "the |
| + // bucket retired it" — a transient Get failure disables removal |
| + // for the whole pass, same as an unreadable account manifest. |
| + readFailed = true |
| + continue |
| + } |
| + rec, err := DecodeRecord(raw) |
| + if err != nil { |
| + log.Printf("provision: skipping record %s: malformed: %v", key, err) |
| + protect[host] = true |
| + continue |
| + } |
| + if rec.Host != host || rec.Account != acct || rec.App != app { |
| + log.Printf("provision: skipping record %s: disagrees with its own path (host=%q account=%q app=%q)", |
| + key, rec.Host, rec.Account, rec.App) |
| + protect[host] = true |
| + continue |
| + } |
| + // The record IS the authorization, and this pass turns its host into |
| + // a socket path and a database path (DefaultSocket/DefaultDBPath) |
| + // and a registry primary key. The console's create door enforces |
| + // this same shape — but a door is not a guarantee about what is |
| + // already in the bucket: anything with bucket write access, and any |
| + // record written by a different (older, or future) writer, arrives |
| + // here unexamined. Both doors hold the invariant, neither trusts |
| + // the other. Protected like a corrupted body, and for the same |
| + // reason: a record that cannot be used is not a record that was |
| + // retired. |
| + if !ValidHostShape(rec.Host) { |
| + log.Printf("provision: skipping record %s: host %q is not a hostname", key, rec.Host) |
| + protect[host] = true |
| + continue |
| + } |
| + // ValidateShape is the console create door's policy check — but a |
| + // door is not a guarantee about what is already in the bucket (the |
| + // same reasoning as the host-shape check just above). This pass |
| + // materializes whatever it finds, so a record whose Kind/Addr/Unit/ |
| + // Environment combination is malformed must be caught here too, or |
| + // a bad value becomes a live route. Protected like a corrupted |
| + // body: the record is there, it just cannot be used. |
| + if err := ValidateShape(rec); err != nil { |
| + log.Printf("provision: skipping record %s: invalid shape: %v", key, err) |
| + protect[host] = true |
| + continue |
| + } |
| + // ValidateShape deliberately does not inspect Channel (it validates |
| + // shape fields, and Channel is not one), but a channel-less record |
| + // must not become — or converge — a route pinned to no channel: |
| + // that is issue #130's failure mode arriving through a different |
| + // door than the one Task 3 closed. Protected like a corrupted body: |
| + // the record is there, it just cannot be used. |
| + if rec.Channel == "" { |
| + log.Printf("provision: skipping record %s: empty channel", key) |
| + protect[host] = true |
| + continue |
| + } |
| + // Scope. For a home-boxed record this is the region/pool test it |
| + // always was; for a static or public-addr one the region half widens |
| + // to every platform box in the deployment (everywhereClass, above). |
| + // A replica record narrows it: it belongs ONLY to the box whose |
| + // placement is the record's own Region — the placement the create |
| + // door validated as armed — so a replica record minted on the wrong |
| + // region would mint a route with no mirror behind it and no lease |
| + // speaking to it, a process holding live order data nobody |
| + // authorized into that region (the residence rule, #162). Every |
| + // other box skips it exactly like any other out-of-scope record; |
| + // the replica's own box sees Region == cfg.Region through the |
| + // ordinary test above. |
| + if !cfg.inScope(rec.Region, rec.Pool, everywhereClass(rec)) { |
| + continue // another agent's jurisdiction, not this pass's business |
| + } |
| + // Two accounts writing the same host is a collision this pass |
| + // cannot arbitrate silently: whichever record this listing reaches |
| + // FIRST wins the host, and the loser is skipped loudly rather than |
| + // overwriting the winner in the map with no trace of the conflict. |
| + if prev, dup := recs[rec.Host]; dup { |
| + log.Printf("provision: duplicate host %s: keeping %s, skipping %s", rec.Host, |
| + RecordKey(prev.Account, prev.App, prev.Host), RecordKey(rec.Account, rec.App, rec.Host)) |
| + continue |
| + } |
| + recs[rec.Host] = rec |
| } |
| - recs[rec.Host] = rec |
| } |
| // --- manifest seed cache, per (acct, app), lazily ------------------------- |
| @@ -822,6 +828,58 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| // never safe to feed the converge branch below — SetSeed-ing an existing |
| // route to "" on a transient blip would erase a working seed for a |
| // reason that has nothing to do with the manifest actually changing. |
| + // Each app with records costs two reads below (its manifest and its trash |
| + // marker), made lazily, one at a time, inside the record loop. Both are |
| + // read here first, eight at a time, for every app the records name; the |
| + // lookups below then judge exactly what they judged before, from these |
| + // results, and read for themselves only for an app not prefetched (a |
| + // route's app with no record, at removal time). |
| + // The manifest is kept only as the seed it renders to, never whole: the |
| + // lookup below has always reduced it to that at once. |
| + type manifestRead struct { |
| + seed string |
| + err error |
| + } |
| + type tombRead struct { |
| + tomb *release.AppTombstone |
| + err error |
| + } |
| + var recApps []string |
| + appOf := map[string]acctApp{} |
| + for _, rec := range recs { |
| + k := acctApp{rec.Account, rec.App} |
| + id := rec.Account + "/" + rec.App |
| + if _, seen := appOf[id]; !seen { |
| + appOf[id] = k |
| + recApps = append(recApps, id) |
| + } |
| + } |
| + slices.Sort(recApps) |
| + manifests := map[acctApp]manifestRead{} |
| + tombs := map[acctApp]tombRead{} |
| + var prefetchMu sync.Mutex |
| + forEach(recApps, func(id string) { |
| + k := appOf[id] |
| + // Trash first, as the record loop asks: an app in the trash, or one |
| + // whose marker will not read, is never minted for, so its manifest |
| + // was never read and is not read here either. |
| + tomb, terr := release.ReadAppTombstone(ctx, s, k.acct, k.app) |
| + prefetchMu.Lock() |
| + tombs[k] = tombRead{tomb, terr} |
| + prefetchMu.Unlock() |
| + if tomb != nil || terr != nil { |
| + return |
| + } |
| + m, merr := GetManifest(ctx, s, k.acct, k.app) |
| + r := manifestRead{err: merr} |
| + if merr == nil { |
| + r.seed = seedJSON(m) |
| + } |
| + prefetchMu.Lock() |
| + manifests[k] = r |
| + prefetchMu.Unlock() |
| + }) |
| + |
| seedCache := map[acctApp]string{} |
| seedKnown := map[acctApp]bool{} |
| seenManifest := map[acctApp]bool{} |
| @@ -831,7 +889,16 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| return seedCache[k], seedKnown[k] |
| } |
| seenManifest[k] = true |
| - m, err := GetManifest(ctx, s, acct, app) |
| + var rendered string |
| + var err error |
| + if r, ok := manifests[k]; ok { |
| + rendered, err = r.seed, r.err |
| + } else { |
| + var m *Manifest |
| + if m, err = GetManifest(ctx, s, acct, app); err == nil { |
| + rendered = seedJSON(m) |
| + } |
| + } |
| if err != nil { |
| if errors.Is(err, os.ErrNotExist) { |
| seedCache[k], seedKnown[k] = "", true // absent manifest: confirmed no seed |
| @@ -845,7 +912,7 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| readFailed = true |
| return "", false |
| } |
| - seedCache[k], seedKnown[k] = seedJSON(m), true |
| + seedCache[k], seedKnown[k] = rendered, true |
| return seedCache[k], seedKnown[k] |
| } |
| @@ -881,7 +948,13 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| return trashCache[k] |
| } |
| trashSeen[k] = true |
| - tomb, err := release.ReadAppTombstone(ctx, s, acct, app) |
| + var tomb *release.AppTombstone |
| + var err error |
| + if r, ok := tombs[k]; ok { |
| + tomb, err = r.tomb, r.err |
| + } else { |
| + tomb, err = release.ReadAppTombstone(ctx, s, acct, app) |
| + } |
| switch { |
| case err != nil: |
| log.Printf("provision: trash marker for %s/%s unreadable: %v — not minting for this app", acct, app, err) |
| @@ -1193,32 +1266,38 @@ func knownAccounts(ctx context.Context, s release.Store) (known map[string]bool, |
| if err != nil { |
| return nil, false, fmt.Errorf("provision: list accounts: %w", err) |
| } |
| + var manifestKeys []string |
| for _, key := range keys { |
| - sqid, ok := domains.AccountManifestSqid(key) |
| - if !ok { |
| - continue |
| + if _, ok := domains.AccountManifestSqid(key); ok { |
| + manifestKeys = append(manifestKeys, key) |
| } |
| - body, _, err := s.Get(ctx, key) |
| - if err != nil { |
| - log.Printf("provision: skipping %s: read failed: %v", key, err) |
| - readFailed = true |
| - continue |
| - } |
| - var m domains.AccountManifest |
| - if err := json.Unmarshal(body, &m); err != nil || m.ID <= 0 || m.Sqid == "" { |
| - log.Printf("provision: skipping %s: malformed account manifest: %v", key, err) |
| - readFailed = true |
| - continue |
| - } |
| - if m.Sqid != sqid || domains.Sqid(m.ID) != sqid { |
| - log.Printf("provision: skipping %s: sqid %q/%q is not Sqid(%d)=%q", key, sqid, m.Sqid, m.ID, domains.Sqid(m.ID)) |
| - readFailed = true |
| - continue |
| - } |
| - if m.Deleted() { |
| - continue |
| + } |
| + for _, part := range chunks(manifestKeys) { |
| + bodies := getEach(ctx, s, part) |
| + for _, key := range part { |
| + sqid, _ := domains.AccountManifestSqid(key) |
| + body, err := bodies[key].body, bodies[key].err |
| + if err != nil { |
| + log.Printf("provision: skipping %s: read failed: %v", key, err) |
| + readFailed = true |
| + continue |
| + } |
| + var m domains.AccountManifest |
| + if err := json.Unmarshal(body, &m); err != nil || m.ID <= 0 || m.Sqid == "" { |
| + log.Printf("provision: skipping %s: malformed account manifest: %v", key, err) |
| + readFailed = true |
| + continue |
| + } |
| + if m.Sqid != sqid || domains.Sqid(m.ID) != sqid { |
| + log.Printf("provision: skipping %s: sqid %q/%q is not Sqid(%d)=%q", key, sqid, m.Sqid, m.ID, domains.Sqid(m.ID)) |
| + readFailed = true |
| + continue |
| + } |
| + if m.Deleted() { |
| + continue |
| + } |
| + known[sqid] = true |
| } |
| - known[sqid] = true |
| } |
| return known, readFailed, nil |
| } |
| @@ -1313,27 +1392,33 @@ func ListAllRecords(ctx context.Context, s release.Store) ([]Record, error) { |
| if partial { |
| return nil, errors.New("provision: an app's records could not be listed — refusing to report a partial record set") |
| } |
| - var recs []Record |
| + var recordKeys []string |
| for _, key := range keys { |
| - acct, app, host, ok := ParseRecordKey(key, known) |
| - if !ok { |
| - continue |
| - } |
| - raw, _, err := s.Get(ctx, key) |
| - if err != nil { |
| - return nil, fmt.Errorf("provision: read record %s: %w", key, err) |
| - } |
| - rec, err := DecodeRecord(raw) |
| - if err != nil { |
| - log.Printf("provision: skipping record %s: malformed: %v", key, err) |
| - continue |
| + if _, _, _, ok := ParseRecordKey(key, known); ok { |
| + recordKeys = append(recordKeys, key) |
| } |
| - if rec.Host != host || rec.Account != acct || rec.App != app { |
| - log.Printf("provision: skipping record %s: disagrees with its own path (host=%q account=%q app=%q)", |
| - key, rec.Host, rec.Account, rec.App) |
| - continue |
| + } |
| + var recs []Record |
| + for _, part := range chunks(recordKeys) { |
| + bodies := getEach(ctx, s, part) |
| + for _, key := range part { |
| + acct, app, host, _ := ParseRecordKey(key, known) |
| + raw, err := bodies[key].body, bodies[key].err |
| + if err != nil { |
| + return nil, fmt.Errorf("provision: read record %s: %w", key, err) |
| + } |
| + rec, err := DecodeRecord(raw) |
| + if err != nil { |
| + log.Printf("provision: skipping record %s: malformed: %v", key, err) |
| + continue |
| + } |
| + if rec.Host != host || rec.Account != acct || rec.App != app { |
| + log.Printf("provision: skipping record %s: disagrees with its own path (host=%q account=%q app=%q)", |
| + key, rec.Host, rec.Account, rec.App) |
| + continue |
| + } |
| + recs = append(recs, rec) |
| } |
| - recs = append(recs, rec) |
| } |
| return recs, nil |
| } |
| @@ -1426,6 +1511,63 @@ func listAppSubtree(ctx context.Context, s release.Store, known map[string]bool, |
| return out, partial, nil |
| } |
| +// readChunk is how many keys a judging loop prefetches at once: enough to keep |
| +// listWalkers busy, few enough that a large record set is never held in |
| +// memory whole, and that a loop stopping at its first bad read has read only |
| +// a chunk ahead of it. |
| +const readChunk = 64 |
| + |
| +// chunks splits keys into runs of at most readChunk, in order. |
| +func chunks(keys []string) [][]string { |
| + var out [][]string |
| + for len(keys) > readChunk { |
| + out = append(out, keys[:readChunk]) |
| + keys = keys[readChunk:] |
| + } |
| + if len(keys) > 0 { |
| + out = append(out, keys) |
| + } |
| + return out |
| +} |
| + |
| +// fetched is one prefetched object: its body, or why it could not be read. |
| +type fetched struct { |
| + body []byte |
| + err error |
| +} |
| + |
| +// getEach reads every key in keys, listWalkers at a time, and returns each |
| +// one's body or error by key. |
| +// |
| +// The record readers used to Get each record in turn inside the loop that |
| +// judged it. With 152 records on the flagship that was the bulk of a 12.5s |
| +// reconcile and a 6.3s moves pass on every serial bump (#231, measured on the |
| +// box with the fast loop's stage timing). The loops are unchanged: they still |
| +// visit keys in the same order and judge each one exactly as before, only |
| +// reading the body from here rather than from the store. A key's read error |
| +// is that key's, as it was. |
| +// |
| +// Callers prefetch one chunk (chunks) at a time and judge it before the next. |
| +// A context already cancelled reads nothing: every key reports ctx.Err(), so |
| +// a judging loop sees a read failure for it, never a missing body. |
| +func getEach(ctx context.Context, s release.Store, keys []string) map[string]fetched { |
| + out := make(map[string]fetched, len(keys)) |
| + if err := ctx.Err(); err != nil { |
| + for _, k := range keys { |
| + out[k] = fetched{err: err} |
| + } |
| + return out |
| + } |
| + var mu sync.Mutex |
| + forEach(keys, func(k string) { |
| + body, _, err := s.Get(ctx, k) |
| + mu.Lock() |
| + out[k] = fetched{body: body, err: err} |
| + mu.Unlock() |
| + }) |
| + return out |
| +} |
| + |
| // forEach runs f over items, listWalkers at a time, and returns when all |
| // have finished. |
| func forEach(items []string, f func(string)) { |
diff --git a/internal/provision/sidecar_reconcile.go b/internal/provision/sidecar_reconcile.go| index 5870d3969..1aeabd4d5 100644 |
| --- a/internal/provision/sidecar_reconcile.go |
| +++ b/internal/provision/sidecar_reconcile.go |
| @@ -255,94 +255,100 @@ func ReconcileSidecars(ctx context.Context, s release.Store, reg *registry.Regis |
| // then delete every sidecar in scope. Narrower than readFailed (which |
| // vetoes removal for the whole pass): only the unusable host is held back. |
| protect := map[string]bool{} |
| + var recordKeys []string |
| for _, key := range keys { |
| - acct, app, name, ok := ParseSidecarKey(key, known) |
| - if !ok { |
| - continue |
| - } |
| - // Derived from the KEY, never from the body: the body is what is |
| - // under suspicion in every branch below. |
| - host := SidecarHost(name, app, acct) |
| - raw, _, err := s.Get(ctx, key) |
| - if err != nil { |
| - log.Printf("provision: skipping sidecar record %s: read failed: %v", key, err) |
| - // Its absence from recs below must not read as "the bucket retired |
| - // it" — a transient Get failure disables removal for the whole |
| - // pass, same as an unreadable account manifest. |
| - readFailed = true |
| - continue |
| + if _, _, _, ok := ParseSidecarKey(key, known); ok { |
| + recordKeys = append(recordKeys, key) |
| } |
| - var rec SidecarRecord |
| - if err := json.Unmarshal(raw, &rec); err != nil { |
| - log.Printf("provision: skipping sidecar record %s: malformed: %v", key, err) |
| - protect[host] = true |
| - continue |
| - } |
| - if rec.Name != name || rec.Account != acct || rec.App != app { |
| - log.Printf("provision: skipping sidecar record %s: disagrees with its own path (name=%q account=%q app=%q)", |
| - key, rec.Name, rec.Account, rec.App) |
| - protect[host] = true |
| - continue |
| - } |
| - // ValidateSidecarShape is the console create door's policy check — but |
| - // a door is not a guarantee about what is already in the bucket: |
| - // anything with bucket write access, and any record written by a |
| - // different (older, or future) writer, arrives here unexamined. Both |
| - // doors hold the invariant, neither trusts the other. Protected like a |
| - // corrupted body, and for the same reason: a record that cannot be |
| - // used is not a record that was retired. |
| - if err := ValidateSidecarShape(rec); err != nil { |
| - log.Printf("provision: skipping sidecar record %s: invalid shape: %v", key, err) |
| - protect[host] = true |
| - continue |
| - } |
| - // The host is DERIVED from the identity, never taken on trust: it |
| - // becomes a database path and a registry primary key, and a record |
| - // free to state its own host could name any row on the box. |
| - if rec.Host != host { |
| - log.Printf("provision: skipping sidecar record %s: host %q is not %q", key, rec.Host, host) |
| - protect[host] = true |
| - continue |
| - } |
| - // ValidateSidecarShape deliberately does not inspect Channel (it |
| - // validates shape fields, and Channel is not one), but a channel-less |
| - // record must not become — or converge — a route pinned to no channel: |
| - // such a row resolves no release, so it has no build to exec, and the |
| - // worker boot pass would try to run it against a symlink that does not |
| - // exist. The console stamps a channel at create, so a record without |
| - // one is malformed. Protected like a corrupted body: the record is |
| - // there, it just cannot be used. |
| - if rec.Channel == "" { |
| - log.Printf("provision: skipping sidecar record %s: empty channel", key) |
| - protect[host] = true |
| - continue |
| - } |
| - if rec.Region != cfg.Region || !poolServed(cfg.Pools, rec.Pool) { |
| - continue // another agent's jurisdiction, not this pass's business |
| - } |
| - // Placement is the third scope axis (design §8), and it is applied |
| - // HERE, beside region and pool, precisely so it behaves like them: a |
| - // record this box may not run never enters recs, and the removal phase |
| - // below therefore evacuates any row it left behind. "none" is a |
| - // property of the BOX's network; "no-inbound" is a requirement of the |
| - // WORKLOAD — the point of the two halves being independent is that a |
| - // member gets a mechanical guarantee rather than an operator's promise, |
| - // which a row left running on an inbound box would not be. |
| - if m, _ := manifestFor(acct, app); !placeable(rec, m, cfg) { |
| - log.Printf("provision: sidecar %s is not placeable here: it asks for %q placement and this box accepts inbound work (CARLOS_INBOUND=%q)", |
| - rec.Host, sidecarPlacement(rec, m), cfg.Inbound) |
| - continue |
| - } |
| - // Two records claiming one host is a collision this pass cannot |
| - // arbitrate silently: whichever this listing reaches FIRST wins, and |
| - // the loser is skipped loudly rather than overwriting the winner in |
| - // the map with no trace of the conflict. |
| - if prev, dup := recs[rec.Host]; dup { |
| - log.Printf("provision: duplicate sidecar host %s: keeping %s, skipping %s", rec.Host, |
| - SidecarRecordKey(prev.Account, prev.App, prev.Name), key) |
| - continue |
| + } |
| + for _, part := range chunks(recordKeys) { |
| + bodies := getEach(ctx, s, part) |
| + for _, key := range part { |
| + acct, app, name, _ := ParseSidecarKey(key, known) |
| + // Derived from the KEY, never from the body: the body is what is |
| + // under suspicion in every branch below. |
| + host := SidecarHost(name, app, acct) |
| + raw, err := bodies[key].body, bodies[key].err |
| + if err != nil { |
| + log.Printf("provision: skipping sidecar record %s: read failed: %v", key, err) |
| + // Its absence from recs below must not read as "the bucket retired |
| + // it" — a transient Get failure disables removal for the whole |
| + // pass, same as an unreadable account manifest. |
| + readFailed = true |
| + continue |
| + } |
| + var rec SidecarRecord |
| + if err := json.Unmarshal(raw, &rec); err != nil { |
| + log.Printf("provision: skipping sidecar record %s: malformed: %v", key, err) |
| + protect[host] = true |
| + continue |
| + } |
| + if rec.Name != name || rec.Account != acct || rec.App != app { |
| + log.Printf("provision: skipping sidecar record %s: disagrees with its own path (name=%q account=%q app=%q)", |
| + key, rec.Name, rec.Account, rec.App) |
| + protect[host] = true |
| + continue |
| + } |
| + // ValidateSidecarShape is the console create door's policy check — but |
| + // a door is not a guarantee about what is already in the bucket: |
| + // anything with bucket write access, and any record written by a |
| + // different (older, or future) writer, arrives here unexamined. Both |
| + // doors hold the invariant, neither trusts the other. Protected like a |
| + // corrupted body, and for the same reason: a record that cannot be |
| + // used is not a record that was retired. |
| + if err := ValidateSidecarShape(rec); err != nil { |
| + log.Printf("provision: skipping sidecar record %s: invalid shape: %v", key, err) |
| + protect[host] = true |
| + continue |
| + } |
| + // The host is DERIVED from the identity, never taken on trust: it |
| + // becomes a database path and a registry primary key, and a record |
| + // free to state its own host could name any row on the box. |
| + if rec.Host != host { |
| + log.Printf("provision: skipping sidecar record %s: host %q is not %q", key, rec.Host, host) |
| + protect[host] = true |
| + continue |
| + } |
| + // ValidateSidecarShape deliberately does not inspect Channel (it |
| + // validates shape fields, and Channel is not one), but a channel-less |
| + // record must not become — or converge — a route pinned to no channel: |
| + // such a row resolves no release, so it has no build to exec, and the |
| + // worker boot pass would try to run it against a symlink that does not |
| + // exist. The console stamps a channel at create, so a record without |
| + // one is malformed. Protected like a corrupted body: the record is |
| + // there, it just cannot be used. |
| + if rec.Channel == "" { |
| + log.Printf("provision: skipping sidecar record %s: empty channel", key) |
| + protect[host] = true |
| + continue |
| + } |
| + if rec.Region != cfg.Region || !poolServed(cfg.Pools, rec.Pool) { |
| + continue // another agent's jurisdiction, not this pass's business |
| + } |
| + // Placement is the third scope axis (design §8), and it is applied |
| + // HERE, beside region and pool, precisely so it behaves like them: a |
| + // record this box may not run never enters recs, and the removal phase |
| + // below therefore evacuates any row it left behind. "none" is a |
| + // property of the BOX's network; "no-inbound" is a requirement of the |
| + // WORKLOAD — the point of the two halves being independent is that a |
| + // member gets a mechanical guarantee rather than an operator's promise, |
| + // which a row left running on an inbound box would not be. |
| + if m, _ := manifestFor(acct, app); !placeable(rec, m, cfg) { |
| + log.Printf("provision: sidecar %s is not placeable here: it asks for %q placement and this box accepts inbound work (CARLOS_INBOUND=%q)", |
| + rec.Host, sidecarPlacement(rec, m), cfg.Inbound) |
| + continue |
| + } |
| + // Two records claiming one host is a collision this pass cannot |
| + // arbitrate silently: whichever this listing reaches FIRST wins, and |
| + // the loser is skipped loudly rather than overwriting the winner in |
| + // the map with no trace of the conflict. |
| + if prev, dup := recs[rec.Host]; dup { |
| + log.Printf("provision: duplicate sidecar host %s: keeping %s, skipping %s", rec.Host, |
| + SidecarRecordKey(prev.Account, prev.App, prev.Name), key) |
| + continue |
| + } |
| + recs[rec.Host] = rec |
| } |
| - recs[rec.Host] = rec |
| } |
| trashed := trashGate(ctx, s, "sidecar") |