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)

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

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

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)