Wait for the assigner's composition only while the assigner is online, say at once that it is not judged when it is not, and take it in when the assigner writes it later, so no check waits its bound for a composition nobody makes (issue 1297, B3d)

2026-10-23 15:44:00 +02:00
parent c9ee92a4ab
commit 7cbb1c4207
8 changed files with 412 additions and 37 deletions
@@ -2,44 +2,98 @@ package main

 import (
 	"context"
-	"errors"
-	"fmt"

 	"github.com/nats-io/nats.go/jetstream"

 	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-checker/internal/check"
 )

-// compositionWatch reads the assigner's composition of a check (ADR 0528 clause 5) by watching its key: no poll and no
-// timer, the check's own bound ends the wait.
-type compositionWatch struct{ kv jetstream.KeyValue }
+// compositionWatch reads the assigner's composition of a check (ADR 0528 clause 5) by watching its key and the
+// assigner's presence: no poll and no timer. With no assigner online it answers check.ErrAssignerNotRunning at once.
+type compositionWatch struct {
+	kv jetstream.KeyValue
+	// presence is the watchman's presence bucket; nil (not there) counts as no assigner online.
+	presence jetstream.KeyValue
+}

-// Await answers the composition under key once the assigner wrote one that does not wait; a value it cannot read is an
-// error, and ctx ending is ctx's error.
+// Await answers the composition under key once the assigner wrote one that does not wait (check.AwaitComposition).
 func (c compositionWatch) Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) {
-	w, err := c.kv.Watch(ctx, key)
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	comps := make(chan check.CompositionEvent)
+	presence := make(chan check.PresenceEvent)
+	cw, err := c.kv.Watch(ctx, key)
 	if err != nil {
 		return delivery.CompositionVerdict{}, err
 	}
-	defer w.Stop()
-	for {
-		select {
-		case <-ctx.Done():
-			return delivery.CompositionVerdict{}, ctx.Err()
-		case e, ok := <-w.Updates():
-			if !ok {
-				return delivery.CompositionVerdict{}, errors.New("the watch of the composition ended")
+	defer cw.Stop()
+	go func() {
+		for e := range cw.Updates() {
+			ev := check.CompositionEvent{Settled: e == nil}
+			if e != nil && e.Operation() == jetstream.KeyValuePut {
+				ev.Value = e.Value()
 			}
+			select {
+			case comps <- ev:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	if c.presence == nil {
+		go func() {
+			select {
+			case presence <- check.PresenceEvent{Settled: true}:
+			case <-ctx.Done():
+			}
+		}()
+	} else {
+		pw, err := c.presence.Watch(ctx, "part."+delivery.AssignerModule+".*")
+		if err != nil {
+			return delivery.CompositionVerdict{}, err
+		}
+		defer pw.Stop()
+		go func() {
+			for e := range pw.Updates() {
+				ev := check.PresenceEvent{Settled: e == nil}
+				if e != nil {
+					ev.Key = e.Key()
+					if e.Operation() == jetstream.KeyValuePut {
+						if p, err := delivery.DecodePresence(e.Value()); err == nil {
+							ev.Online = p.Online
+						}
+					}
+				}
+				select {
+				case presence <- ev:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}()
+	}
+	return check.AwaitComposition(ctx, presence, comps)
+}
+
+// followCompositions hands the checker each composition the assigner writes from now on, so a verdict that said its
+// composition was not judged takes it in when the assigner starts (event-driven).
+func followCompositions(ctx context.Context, kv jetstream.KeyValue, ch *check.Checker) error {
+	w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
+	if err != nil {
+		return err
+	}
+	go func() {
+		defer w.Stop()
+		for e := range w.Updates() {
 			if e == nil || e.Operation() != jetstream.KeyValuePut {
 				continue
 			}
-			v, err := delivery.DecodeCompositionVerdict(e.Value())
-			if err != nil {
-				return delivery.CompositionVerdict{}, fmt.Errorf("the composition %s cannot be read: %w", key, err)
-			}
-			if v.State != delivery.CompositionWaits {
-				return v, nil
+			if comp, err := delivery.DecodeCompositionVerdict(e.Value()); err == nil {
+				ch.Composed(ctx, comp)
 			}
 		}
-	}
+	}()
+	return nil
 }
@@ -4,6 +4,7 @@ package main

 import (
 	"context"
+	"errors"
 	"os"
 	"path/filepath"
 	"testing"
@@ -14,6 +15,8 @@ import (
 	"github.com/nats-io/nats.go/jetstream"

 	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-checker/internal/check"
 )

 // 1297/AC3 (its bus half): on the real bus, the checker's read of the assigner's composition waits while the key is
@@ -45,10 +48,31 @@ func TestTheCheckerAwaitsTheAssignersComposition(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	pkv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.PresenceBucket, History: 5})
+	if err != nil {
+		t.Fatal(err)
+	}
+	pkey, _ := delivery.PresencePartKey(delivery.AssignerModule, "novox")
+	on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
+	if _, err := pkv.Put(ctx, pkey, on); err != nil {
+		t.Fatal(err)
+	}
+	// With no assigner online the wait answers at once (a second key, before the assigner of this test is online).
+	off, _ := delivery.EncodePresence(delivery.Presence{Online: false})
+	if _, err := pkv.Put(ctx, pkey, off); err != nil {
+		t.Fatal(err)
+	}
+	other, _ := delivery.CompositionKey("feat/y", "0123456789ab")
+	if _, err := (compositionWatch{kv: kv, presence: pkv}).Await(ctx, other); !errors.Is(err, check.ErrAssignerNotRunning) {
+		t.Fatalf("with no assigner online the wait answered %v", err)
+	}
+	if _, err := pkv.Put(ctx, pkey, on); err != nil {
+		t.Fatal(err)
+	}
 	key, _ := delivery.CompositionKey("feat/x", "0123456789ab")
 	got := make(chan delivery.CompositionVerdict, 1)
 	go func() {
-		v, err := compositionWatch{kv}.Await(ctx, key)
+		v, err := compositionWatch{kv: kv, presence: pkv}.Await(ctx, key)
 		if err != nil {
 			t.Error(err)
 		}
@@ -231,7 +231,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	case err != nil:
 		return fmt.Errorf("the bucket %s: %w", delivery.AssignerCompositionsBucket, err)
 	default:
-		compositions = compositionWatch{kv}
+		cw := compositionWatch{kv: kv}
+		if pkv, err := js.KeyValue(ctx, 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)
+		}
 	}
 	// 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}