amadan
provision: read records eight at a time in the reconcile and record passes
The fast loop's new stage timing, on the flagship's IE box after the 82c5e17b roll: serial pass stages took 27.2s (reconcile 12.8s, telemetry 1.8s, moves 6.3s, aliases 200ms, adopt 3.3s, config 2.8s). The adopt stage that notices a promote waits behind the rest, so a promote took ~27s to be seen. Reconcile and moves (ListAllRecords) each read all 152 instance records one Get at a time. getEach reads the keys a loop will judge, eight at a time, before the loop runs. The loops are unchanged: same order, same judgement, each key's read error still that key's. Applied to Reconcile, ListAllRecords, ReconcileSidecars and the account manifests every reader reads first.
Human
Prompt
Keep investigating overnight until it's sorted.
From this commit's own message. The prompt ledger has no entries for it — which is what a squashed, rebased or amended commit looks like, and what every commit looks like in a repo without a ledger.
Summary
Prefetches instance records, sidecar records and account manifests eight at a time before the reconcile loops judge them, cutting the serial-bump pass that delays promote pickup.
Agent notes
Measured stage times are in the commit body (journal of i-092c0c1eea75723cb, 2026-10-05 22:44-22:45). TestRecordReadersReadRecordsConcurrently fails with listWalkers set to 1.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3 files changed,
+161
−4
internal/provision/parallelreads_test.go+101 −0internal/provision/reconcile.go+52 −3internal/provision/sidecar_reconcile.go+8 −1
CI cancelled — superseded by a newer push — run details
diff --git a/internal/provision/parallelreads_test.go b/internal/provision/parallelreads_test.go| new file mode 100644 |
| index 000000000..e0e049e90 |
| --- /dev/null |
| +++ b/internal/provision/parallelreads_test.go |
| @@ -0,0 +1,101 @@ |
| +package provision |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "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 |
| + failKey string |
| +} |
| + |
| +func (s *overlapStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if strings.Contains(key, "/instances/") { |
| + s.mu.Lock() |
| + s.inFlight++ |
| + 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 { |
| + t.Fatalf("%s read at most %d record at a time, want several", c.name, s.maxFlight) |
| + } |
| + } |
| +} |
| + |
| +// 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) |
| + } |
| + for _, h := range []string{"ra.go.tito.io", "rb.go.tito.io", "rc.go.tito.io", "rd.go.tito.io"} { |
| + if rt, err := reg.Get(h); err != nil || rt == nil { |
| + t.Fatalf("%s gone after a pass where only rb's read failed: %+v, %v", h, rt, err) |
| + } |
| + } |
| + if _, err := ListAllRecords(ctx, s); err == nil { |
| + t.Fatal("ListAllRecords returned a set with one record unread") |
| + } |
| +} |
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go| index b77b5e50e..aaef5f92e 100644 |
| --- a/internal/provision/reconcile.go |
| +++ b/internal/provision/reconcile.go |
| @@ -736,12 +736,19 @@ 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 { |
| + if _, _, _, ok := ParseRecordKey(key, known); ok { |
| + recordKeys = append(recordKeys, key) |
| + } |
| + } |
| + bodies := getEach(ctx, s, recordKeys) |
| for _, key := range keys { |
| acct, app, host, ok := ParseRecordKey(key, known) |
| if !ok { |
| continue |
| } |
| - raw, _, err := s.Get(ctx, key) |
| + 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 |
| @@ -1216,12 +1223,19 @@ 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 { |
| + if _, ok := domains.AccountManifestSqid(key); ok { |
| + manifestKeys = append(manifestKeys, key) |
| + } |
| + } |
| + bodies := getEach(ctx, s, manifestKeys) |
| for _, key := range keys { |
| sqid, ok := domains.AccountManifestSqid(key) |
| if !ok { |
| continue |
| } |
| - body, _, err := s.Get(ctx, key) |
| + body, err := bodies[key].body, bodies[key].err |
| if err != nil { |
| log.Printf("provision: skipping %s: read failed: %v", key, err) |
| readFailed = true |
| @@ -1336,13 +1350,20 @@ 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 recordKeys []string |
| + for _, key := range keys { |
| + if _, _, _, ok := ParseRecordKey(key, known); ok { |
| + recordKeys = append(recordKeys, key) |
| + } |
| + } |
| + bodies := getEach(ctx, s, recordKeys) |
| var recs []Record |
| for _, key := range keys { |
| acct, app, host, ok := ParseRecordKey(key, known) |
| if !ok { |
| continue |
| } |
| - raw, _, err := s.Get(ctx, key) |
| + raw, err := bodies[key].body, bodies[key].err |
| if err != nil { |
| return nil, fmt.Errorf("provision: read record %s: %w", key, err) |
| } |
| @@ -1449,6 +1470,34 @@ func listAppSubtree(ctx context.Context, s release.Store, known map[string]bool, |
| return out, partial, nil |
| } |
| +// 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. |
| +func getEach(ctx context.Context, s release.Store, keys []string) map[string]fetched { |
| + out := make(map[string]fetched, len(keys)) |
| + 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 b8ee42868..a861f7cfe 100644 |
| --- a/internal/provision/sidecar_reconcile.go |
| +++ b/internal/provision/sidecar_reconcile.go |
| @@ -255,6 +255,13 @@ 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 { |
| + if _, _, _, ok := ParseSidecarKey(key, known); ok { |
| + recordKeys = append(recordKeys, key) |
| + } |
| + } |
| + bodies := getEach(ctx, s, recordKeys) |
| for _, key := range keys { |
| acct, app, name, ok := ParseSidecarKey(key, known) |
| if !ok { |
| @@ -263,7 +270,7 @@ func ReconcileSidecars(ctx context.Context, s release.Store, reg *registry.Regis |
| // 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) |
| + 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 |