From d0354574c69a034a69a8f0ce6b2b5d00853968c1 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 16:12:18 +0200 Subject: [PATCH] Look the assigner's compositions bucket up at each check until it is there, so an assigner delivered after the checker started is read without a restart (issue 1297, review round 1 item 3) --- .../cmd/mesh-delivery-checker/compositions.go | 32 +++++++++++++ .../compositions_test.go | 46 +++++++++++++++++++ .../cmd/mesh-delivery-checker/main.go | 25 +++++----- 3 files changed, 90 insertions(+), 13 deletions(-) create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_test.go diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go index 5770315..c5a3fe3 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go @@ -2,6 +2,8 @@ package main import ( "context" + "errors" + "sync" "github.com/nats-io/nats.go/jetstream" @@ -97,3 +99,33 @@ func followCompositions(ctx context.Context, kv jetstream.KeyValue, ch *check.Ch }() return nil } + +// lazyCompositions opens the assigner's compositions when a check first asks for one and the bucket is there: until it +// is, no assigner ever ran, so the check says at once that its composition is not judged (check.ErrAssignerNotRunning), +// and the next check looks again. No timer: each look is a check asking. +type lazyCompositions struct { + open func(ctx context.Context) (check.CompositionReader, error) + + mu sync.Mutex + reader check.CompositionReader +} + +// Await is the opened reader's Await, opening it first when it is not yet. +func (l *lazyCompositions) Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) { + l.mu.Lock() + r := l.reader + if r == nil { + opened, err := l.open(ctx) + switch { + case errors.Is(err, jetstream.ErrBucketNotFound): + l.mu.Unlock() + return delivery.CompositionVerdict{}, check.ErrAssignerNotRunning + case err != nil: + l.mu.Unlock() + return delivery.CompositionVerdict{}, err + } + l.reader, r = opened, opened + } + l.mu.Unlock() + return r.Await(ctx, key) +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_test.go new file mode 100644 index 0000000..64afd0b --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_test.go @@ -0,0 +1,46 @@ +package main + +import ( + "context" + "errors" + "testing" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-checker/internal/check" +) + +type answering struct{ v delivery.CompositionVerdict } + +func (a answering) Await(context.Context, string) (delivery.CompositionVerdict, error) { + return a.v, nil +} + +// 1297 review item 3: the compositions bucket missing when the checker starts is looked up again at the next check: while +// it is missing a check says at once the assigner is not running; once it is there the check reads its composition, and +// it is opened once. +func TestTheCompositionsBucketIsLookedUpAtEachCheck(t *testing.T) { + opens, there := 0, false + l := &lazyCompositions{open: func(context.Context) (check.CompositionReader, error) { + opens++ + if !there { + return nil, jetstream.ErrBucketNotFound + } + return answering{delivery.CompositionVerdict{State: delivery.CompositionPass, SDK: "x"}}, nil + }} + if _, err := l.Await(context.Background(), "k"); !errors.Is(err, check.ErrAssignerNotRunning) { + t.Fatalf("without the bucket: %v", err) + } + there = true + for range 2 { + v, err := l.Await(context.Background(), "k") + if err != nil || v.State != delivery.CompositionPass { + t.Fatalf("with the bucket: %+v, %v", v, err) + } + } + if opens != 2 { + t.Errorf("opened %d times, want twice (missing, then once)", opens) + } +} 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 08b3c0b..e3f6669 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -222,24 +222,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro default: builds = buildReader{kvStore{kv}} } - // The assigner's compositions (ADR 0528 clause 5, issue 1297); where its bucket is not there yet a verdict says the - // composition was not run, until the checker starts again. - var compositions check.CompositionReader - switch kv, err := js.KeyValue(ctx, delivery.AssignerCompositionsBucket); { - case errors.Is(err, jetstream.ErrBucketNotFound): - logger.Printf("the bucket %s does not exist yet: a verdict says the composition was not run until the checker starts again", delivery.AssignerCompositionsBucket) - case err != nil: - return fmt.Errorf("the bucket %s: %w", delivery.AssignerCompositionsBucket, err) - default: + // The assigner's compositions (ADR 0528 clause 5, issue 1297): their bucket is looked up at each check until it is + // there, so an assigner delivered after the checker started is read at the next check, and followed from then on. + compositions := &lazyCompositions{open: func(at context.Context) (check.CompositionReader, error) { + kv, err := js.KeyValue(at, delivery.AssignerCompositionsBucket) + if err != nil { + return nil, err + } cw := compositionWatch{kv: kv} - if pkv, err := js.KeyValue(ctx, delivery.PresenceBucket); err == nil { + if pkv, err := js.KeyValue(at, delivery.PresenceBucket); err == nil { cw.presence = pkv } - compositions = cw if err := followCompositions(ctx, kv, ch); err != nil { - return fmt.Errorf("the bucket %s cannot be watched: %w", delivery.AssignerCompositionsBucket, err) + return nil, fmt.Errorf("the bucket %s cannot be watched: %w", delivery.AssignerCompositionsBucket, err) } - } + logger.Printf("the assigner's compositions are read from %s", delivery.AssignerCompositionsBucket) + return cw, nil + }} // The tries (ADR 0501 ยง5) are judged by the same judge, held by the same deliveries. tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesBucket]}, Bound: checkBound, Logf: logger.Printf} holds := newHoldWatch(ctx, js, func(ctx context.Context) { ch.Reconsider(ctx); tryer.Reconsider(ctx) }, logger)