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

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,

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)

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)
 		}
 	}