diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 7d71872..15f126c 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -154,7 +154,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Heard in the background, and again at a reconnect or an ask while it is not; whether it is heard is kept in the // state hearing, which builder_stuck says (dead_letters). dead := &buildrun.DeadLetters{} - hearingKV := process.OptionalBucket(ctx, js, delivery.BuilderModule+"_"+buildrun.HearingState, "builder_stuck cannot say whether DEAD_LETTERS is heard", logger.Printf) + // The state may not be on the bus yet when a delivery started the builder (the grants part creates it, issue 1790): it + // is bound when the bus says it was created, and the latest hearing is written then. + hearingKV := process.AwaitBucket(ctx, js, conn, delivery.BuilderModule+"_"+buildrun.HearingState, "builder_stuck cannot say whether DEAD_LETTERS is heard", logger.Printf) hearing := buildrun.NewDeadLetterHearing(ctx, buildrun.JetStreamDeadLetters{JS: js}, delivery.AskLanguages, dead, logger.Printf, buildrun.SayHearing(ctx, hearingKV, logger.Printf)) defer hearing.Stop() @@ -290,24 +292,25 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // A person asks a key's build again with builder_ask_again, which writes the reasks state (issue 1945): watched, as // every other input. The builder asks again by itself without it. - if kv := process.OptionalBucket(ctx, js, build.ReAsksBucket, "builder_ask_again is not heard until the builder starts again; an ask that can no longer be answered is still asked again by itself", logger.Printf); kv != nil { + // A state not on the bus yet at this start is watched once it appears (issue 1790). + process.AwaitBucket(ctx, js, conn, build.ReAsksBucket, "builder_ask_again is not heard; an ask that can no longer be answered is still asked again by itself", logger.Printf).Ready(func(kv jetstream.KeyValue) { eachReAsk := func(e jetstream.KeyValueEntry) { reAsked(ctx, b, kv, e, logger.Printf) } if rw, err := settle(ctx, kv, eachReAsk); err != nil { process.NotWatched(logger.Printf, build.ReAsksBucket, err, "builder_ask_again is not heard until the builder starts again") } else { go func() { errs <- follow(ctx, rw, eachReAsk) }() } - } + }) // A person reads a build seat's stream with builder_seat_queue, which writes the queues state (issue 1945 note 62): the // builder alone may read the stream, so it answers there. - if kv := process.OptionalBucket(ctx, js, build.QueuesBucket, "builder_seat_queue is not answered until the builder starts again", logger.Printf); kv != nil { + process.AwaitBucket(ctx, js, conn, build.QueuesBucket, "builder_seat_queue is not answered", logger.Printf).Ready(func(kv jetstream.KeyValue) { eachRead := func(e jetstream.KeyValueEntry) { queueRead(ctx, heldOnTheBus{js}, kv, e, logger.Printf) } if qw, err := settle(ctx, kv, eachRead); err != nil { process.NotWatched(logger.Printf, build.QueuesBucket, err, "builder_seat_queue is not answered until the builder starts again") } else { go func() { errs <- follow(ctx, qw, eachRead) }() } - } + }) logger.Printf("started on %s", cred.Node) go func() { errs <- follow(ctx, w, each) }() select { diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index b206980..3c79bbe 100644 --- a/modules/mesh-delivery-builder/go.mod +++ b/modules/mesh-delivery-builder/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-builder go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-builder/go.sum b/modules/mesh-delivery-builder/go.sum index 9a6ce0a..e793069 100644 --- a/modules/mesh-delivery-builder/go.sum +++ b/modules/mesh-delivery-builder/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530 h1:NwCb+gOh0UYqlyG6hG2/jffE/Gd0c7YWd4iJlaCB0lU= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 h1:ikD/ggf9/5k23sxJOX1PFcGzb4M3+QeC66AT0MyQ9Zs= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index f403fe8..767a70e 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -205,7 +205,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Heard in the background, and again at a reconnect or an ask while it is not; whether it is heard is kept in the // state hearing, which checker_status says (dead_letters). dead := &buildrun.DeadLetters{} - hearingKV := process.OptionalBucket(ctx, js, check.HearingBucket, "checker_status cannot say whether DEAD_LETTERS is heard", logger.Printf) + // The state may not be on the bus yet when a delivery started the checker (the grants part creates it, issue 1790): it + // is bound when the bus says it was created, and the latest hearing is written then. + hearingKV := process.AwaitBucket(ctx, js, conn, check.HearingBucket, "checker_status cannot say whether DEAD_LETTERS is heard", logger.Printf) hearing := buildrun.NewDeadLetterHearing(ctx, buildrun.JetStreamDeadLetters{JS: js}, []string{delivery.AskGo, delivery.AskNode}, dead, logger.Printf, buildrun.SayHearing(ctx, hearingKV, logger.Printf)) defer hearing.Stop() diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index 012f639..4e2c71e 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/go.sum b/modules/mesh-delivery-checker/go.sum index 9a6ce0a..e793069 100644 --- a/modules/mesh-delivery-checker/go.sum +++ b/modules/mesh-delivery-checker/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530 h1:NwCb+gOh0UYqlyG6hG2/jffE/Gd0c7YWd4iJlaCB0lU= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 h1:ikD/ggf9/5k23sxJOX1PFcGzb4M3+QeC66AT0MyQ9Zs= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= 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 672e4e6..0b4258d 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -363,14 +363,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // A person takes a registered module in now with indexer_take_in, which writes the takeins state (issue 1945): watched, // as every other input. A module the forge could not read is also read again by itself on the next source revision and // merged message (index.RetryUnread). - if kv := process.OptionalBucket(ctx, js, index.TakeInsBucket, "indexer_take_in is not answered until the indexer starts again; a module the forge could not read is still read again by itself", logger.Printf); kv != nil { + // A state not on the bus yet at this start is watched once it appears (issue 1790). + process.AwaitBucket(ctx, js, conn, index.TakeInsBucket, "indexer_take_in is not answered; a module the forge could not read is still read again by itself", logger.Printf).Ready(func(kv jetstream.KeyValue) { eachTakeIn := func(e jetstream.KeyValueEntry) { takenIn(ctx, ix, kv, e, logger.Printf) } if tw, err := settle(ctx, kv, eachTakeIn); err != nil { process.NotWatched(logger.Printf, index.TakeInsBucket, err, "indexer_take_in is not answered until the indexer starts again") } else { go func() { errs <- follow(ctx, tw, eachTakeIn) }() } - } + }) // The modules the trunk no longer holds go from the bucket (issue 1900); a merge does the same from its own message. if err := ix.Reconcile(ctx); err != nil { logger.Printf("dropping the modules the trunk does not hold at start: %v", err) diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index e8728ce..5a8d354 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/go.sum b/modules/mesh-delivery-indexer/go.sum index 8e94e90..e793069 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -1,7 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 h1:ikD/ggf9/5k23sxJOX1PFcGzb4M3+QeC66AT0MyQ9Zs= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-grants/go.mod b/modules/mesh-grants/go.mod index 76ac8cc..6d8f4e9 100644 --- a/modules/mesh-grants/go.mod +++ b/modules/mesh-grants/go.mod @@ -3,7 +3,7 @@ module mesh-grants go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-grants/go.sum b/modules/mesh-grants/go.sum index 78176f6..4cc1a88 100644 --- a/modules/mesh-grants/go.sum +++ b/modules/mesh-grants/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530 h1:NwCb+gOh0UYqlyG6hG2/jffE/Gd0c7YWd4iJlaCB0lU= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 h1:ikD/ggf9/5k23sxJOX1PFcGzb4M3+QeC66AT0MyQ9Zs= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=