From efdfe1f755186fce84109a6172d23fd93ce0c665 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 15:43:52 +0100 Subject: [PATCH] A bucket the bus cannot reach is its error, not nothing published; only a bucket not found is not yet (issue 1966, review round 3) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-indexer/main.go | 32 +++++++++++++------ .../cmd/mesh-delivery-indexer/main_test.go | 21 ++++++++++++ .../internal/index/packages.go | 2 +- 3 files changed, 44 insertions(+), 11 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index ddfa67b1..00ef9a3d 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -52,7 +52,14 @@ func main() { // awaitedStore is a state bound when it appears (process.AwaitBucket) as the indexer's store: until then every call is // process.ErrNotYet. -type awaitedStore struct{ b *process.AwaitedBucket } +type awaitedStore struct { + b *process.AwaitedBucket + // js and name, when set, tell a bucket that does not exist from one that cannot be reached: the awaited bucket is nil for + // both (the SDK drops the bus's error), and only the first is "not yet". The packages state is read this way: an unreached + // state is no package published, and the version line written over every import form (issue 1966). + js process.BucketOpener + name string +} // expiringKeys is index.Bounds on a state whose keys take a time to live (key-ttl in the manifest): a key armed expires // at its end, which its watcher hears as a purge; one standing keeps its end. @@ -88,6 +95,11 @@ func (e expiringKeys) Disarm(ctx context.Context, key string) error { func (s awaitedStore) store() (kvStore, error) { kv := s.b.KV() if kv == nil { + if s.js != nil { + if _, err := s.js.KeyValue(context.Background(), s.name); err != nil && !errors.Is(err, jetstream.ErrBucketNotFound) { + return kvStore{}, fmt.Errorf("the bucket %s cannot be reached: %w", s.name, err) + } + } return kvStore{}, process.ErrNotYet } return kvStore{kv}, nil @@ -411,17 +423,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro seatBus := &seatBus{defs: awaitedSource{ctx, seatDefs}, marks: awaitedSource{ctx, marks}} seatReader, saySeats := followSeatList(seatBus, saySeatDefinition(js, cred), logger.Printf) readWhenBound(seatReader, seatDefs, marks) - ix := &index.Indexer{Findings: awaitedStore{findings}, OldSections: awaitedStore{oldSections}, Requirements: awaitedStore{requirements}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, - Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(conn, cred, logger.Printf), SayAccepted: sayAccepted(conn, cred), - Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{configCandidates}, Logf: logger.Printf, + ix := &index.Indexer{Findings: awaitedStore{b: findings}, OldSections: awaitedStore{b: oldSections}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known, + Verdicts: awaitedStore{b: verdicts}, SayRefused: sayRefused(conn, cred, logger.Printf), SayAccepted: sayAccepted(conn, cred), + Predefined: awaitedStore{b: predefined}, ConfigCandidates: awaitedStore{b: configCandidates}, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), - ReadOpen: openVia(calls.Call), Packages: awaitedStore{packages}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call), - OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{ownBuildState}, TryRecovery: awaitedStore{tryRecovery}, Graph: awaitedStore{graph}, - SeatDefinitions: awaitedStore{seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats, + ReadOpen: openVia(calls.Call), Packages: awaitedStore{b: packages, js: js, name: index.PackagesBucket}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call), + OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, TryRecovery: awaitedStore{b: tryRecovery}, Graph: awaitedStore{b: graph}, + SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats, Message: messageVia(calls.Call), // The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear; // until then every merge is taken in at once, as before. - Pending: awaitedStore{process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)}, + Pending: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)}, SayTakeIn: sayTakeIn(js, cred)} // Which build this program is, fixed here once, before it hears any merge and never asked again (index.FixOwnBuild): the // node-engine names the successor while it replaces this process, and an old process that learnt it later would take @@ -483,7 +495,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The holds, new with ADR 0632 (issue 1790's kind), bound when they appear: until then a merge that would hold a // manifest is left for redelivery, said, rather than write what a counted engine cannot read. holds := process.AwaitBucket(ctx, js, conn, index.HoldsBucket, "a merge whose manifest every node reads, needing what not every counted node-engine states, is left for redelivery", logger.Printf) - ix.Holds = awaitedStore{holds} + ix.Holds = awaitedStore{b: holds} holds.Ready(func(jetstream.KeyValue) { ix.KickHolds(ctx) }) // Each standing hold's bound, a key that expires at its end (ADR 0632 §2, the indexer's own state): its expiry weighs // the holds again, which raises manifest-held; until the state is bound a hold past its bound is raised at its next @@ -696,7 +708,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro taken.Ready(func(jetstream.KeyValue) { ix.KickFill(ctx) }) // What the indexer heard of FORGE (issue 1126 note 76) is bound when its state appears, as every state a delivery adds // (issue 1790): until then no message is written there, and the rest of the indexer runs. - heard := awaitedStore{process.AwaitBucket(ctx, js, conn, index.HeardBucket, "no FORGE message the indexer handles is written for indexer_heard", logger.Printf)} + heard := awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.HeardBucket, "no FORGE message the indexer handles is written for indexer_heard", logger.Printf)} // Each change of the states its readers read is said on its views family (ADR 0615 §5), from the mark each state was // said up to; a first view (the tools bundle) answers at that mark. startViews(ctx, js, cred, marks, diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index 0c15f040..40e32bfe 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "log" "strings" "testing" @@ -151,8 +152,28 @@ func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part) t.Errorf("forge/group's merged message read as a group's end: %q", logs.String()) } }) + // Issue 1966: a bucket the bus has not made is "not yet"; one it cannot reach (jetstream not enabled, no responders) is the + // bus's own error, which stops a pass instead of reading as nothing published. + t.Run("a bucket not found is not yet, one the bus cannot reach is its error", func(t *testing.T) { + for _, c := range []struct { + name string + err error + notYet bool + }{{"not found", jetstream.ErrBucketNotFound, true}, {"not reachable", jetstream.ErrJetStreamNotEnabled, false}} { + b := process.AwaitBucket(context.Background(), opener{c.err}, nil, "packages", "none", t.Logf) + _, err := awaitedStore{b: b, js: opener{c.err}, name: "packages"}.Keys(context.Background()) + if errors.Is(err, process.ErrNotYet) != c.notYet || err == nil || (!c.notYet && !errors.Is(err, c.err)) { + t.Errorf("%s: Keys answered %v, want not-yet %v", c.name, err, c.notYet) + } + } + }) } +// opener is a bus that answers every bucket with err. +type opener struct{ err error } + +func (o opener) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, o.err } + func TestTheSourceMainKeyIsTakenAtItsRevision(t *testing.T) { // 1344/AC1 (the watch's side: key, revision, value) var logs bytes.Buffer logger := log.New(&logs, "", 0) diff --git a/modules/mesh-delivery-indexer/internal/index/packages.go b/modules/mesh-delivery-indexer/internal/index/packages.go index 09590432..5013979f 100644 --- a/modules/mesh-delivery-indexer/internal/index/packages.go +++ b/modules/mesh-delivery-indexer/internal/index/packages.go @@ -92,7 +92,7 @@ func (ix *Indexer) publishedForPass(ctx context.Context, held map[string]meshcat } var had []string for key, m := range held { - if m.Inputs2 != nil && m.Inputs2.ImportForm != "" { + if m.Inputs2 != nil && m.Inputs2.ImportForm != "" { // not Packages: a version line holds those too, with nothing published had = append(had, key) } }