Say the verdict in a line of at most 140 characters and owe the status mesh/check in the checker's bucket, which the forge bridge performs once per revision; the checker's stream checks wakes the readers of its buckets (issues 1682, 1683)

2026-10-21 21:20:42 +02:00
parent 4242fb5b0b
commit 96ba6bd3dc
19 changed files with 646 additions and 43 deletions
@@ -91,9 +91,34 @@ func (p *performer) refuse(ctx context.Context, doneKey string, rev uint64, a br
 	return true
 }

-// owed performs the act an owed key asks at revision rev, once.
+// owed performs the act an owed key of the gate asks at revision rev, once.
 func (p *performer) owed(ctx context.Context, key string, rev uint64, value []byte) {
-	doneKey := "owed." + key
+	p.perform(ctx, "owed."+key, key, rev, value)
+}
+
+// checkOwed performs the status the checker owes on a head (mesh-delivery-checker_owed; novox/hq ADR 0484 clause 2) at
+// revision rev, once. Its keys are kept apart from the gate's, and a key that is not the checker's one act, the status
+// mesh/check, is left.
+func (p *performer) checkOwed(ctx context.Context, key string, rev uint64, value []byte) {
+	doneKey := "check." + key
+	if p.seen(ctx, doneKey, rev) {
+		return
+	}
+	if _, _, _, err := delivery.ParseCheckerOwedKey(key); err != nil {
+		p.logger.Printf("%s at revision %d is not a status of the checker, and is left: %v", key, rev, err)
+		p.keep(ctx, doneKey, rev)
+		return
+	}
+	if _, err := delivery.DecodeCheckerOwed(value); err != nil {
+		p.logger.Printf("%s at revision %d is not a status of the checker, and is left: %v", key, rev, err)
+		p.keep(ctx, doneKey, rev)
+		return
+	}
+	p.perform(ctx, doneKey, key, rev, value)
+}
+
+// perform does the act of an owed key once for the revision, under the bookkeeping key doneKey.
+func (p *performer) perform(ctx context.Context, doneKey, key string, rev uint64, value []byte) {
 	if p.seen(ctx, doneKey, rev) {
 		return
 	}
@@ -208,3 +208,57 @@ func TestAMergeWithTwoHandedStopsAtTheFirstRefusal(t *testing.T) { // ADR 0441
 		t.Errorf("calls %d messages %d, want the first refusal to stop the pass", len(f.calls), len(pub.data))
 	}
 }
+
+// checkerOwed is the status the checker owes, as it writes it (mesh-sdk go/delivery CheckerOwedKey, EncodeCheckerOwed).
+func checkerOwed(t *testing.T, number int, state, text string) (string, []byte) {
+	t.Helper()
+	key, err := delivery.CheckerOwedKey("novox/mesh-delivery", number, headA)
+	if err != nil {
+		t.Fatal(err)
+	}
+	v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state, Text: text, Target: "https://view.example/checks/g.0123456789ab"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return key, v
+}
+
+func TestTheCheckersStatusIsSetOncePerRevisionAndAKeyOfTheGateIsNotItsKey(t *testing.T) { // 1683/AC5 (ADR 0484 clause 2)
+	f := &forgeFake{}
+	p, _, done, _ := newPerformer(f)
+	key, v := checkerOwed(t, 12, "success", "pass aaaaaaaaaaaa on bbbbbbbbbbbb")
+	p.checkOwed(context.Background(), key, 3, v)
+	p.checkOwed(context.Background(), key, 3, v)
+	if len(f.calls) != 1 || f.calls[0].key != "gitea.gitea_commit_status" ||
+		f.calls[0].body != `{"context":"mesh/check","description":"pass aaaaaaaaaaaa on bbbbbbbbbbbb","owner":"novox","repo":"mesh-delivery","sha":"`+headA+`","state":"success","target_url":"https://view.example/checks/g.0123456789ab"}` {
+		t.Fatalf("calls = %+v", f.calls)
+	}
+	p.checkOwed(context.Background(), key, 4, v)
+	if len(f.calls) != 2 {
+		t.Errorf("a new revision was not performed: %+v", f.calls)
+	}
+	if _, ok := done["check."+key]; !ok {
+		t.Errorf("revision kept under %v, want check.%s", done, key)
+	}
+	// The gate's own owed bookkeeping is another key: the same owed key seen in the gate's bucket is its own act.
+	if _, ok := done["owed."+key]; ok {
+		t.Errorf("the checker's status was kept under the gate's key: %v", done)
+	}
+}
+
+func TestAnythingButTheCheckersStatusInItsBucketIsLeft(t *testing.T) { // ADR 0484 clause 2: the checker owes one act
+	f := &forgeFake{}
+	p, _, done, logs := newPerformer(f)
+	// A comment, and a status of another context: the SDK refuses both to the checker's encoder, so they are written raw.
+	commentKey, commentValue := owedKV(t, "novox/mesh-delivery", 12, "comment.1", delivery.Owed{Act: delivery.OwedComment, Text: "hello"})
+	gateKey, gateValue := owedKV(t, "novox/mesh-delivery", 12, "status=mesh/merge-gate",
+		delivery.Owed{Act: delivery.OwedStatus, Context: "mesh/merge-gate", State: "success", Text: "x", Target: "https://x.example"})
+	p.checkOwed(context.Background(), commentKey, 1, commentValue)
+	p.checkOwed(context.Background(), gateKey, 1, gateValue)
+	if len(f.calls) != 0 {
+		t.Errorf("the forge was called: %+v", f.calls)
+	}
+	if !strings.Contains(logs.String(), commentKey) || !strings.Contains(logs.String(), gateKey) || len(done) != 2 {
+		t.Errorf("log %q, kept %v: both keys are named and kept so they are not tried again", logs.String(), done)
+	}
+}
@@ -30,6 +30,10 @@ import (
 // The gate's buckets mesh-delivery-gate_merges and mesh-delivery-gate_owed are read and watched (the manifest's reads), the
 // bucket performed is the bridge's own (its state), and the five forge tools of the gate's acts are invoked (issue 1341, 1342).
 //
+// The checker's owed bucket is read like the gate's (reads) and the checker's stream checks is heard like EVENTS is (consumes
+// mesh-delivery-checker.owed.written; ADR 0484 clause 2): those entries are written by analogy with the ones the controller
+// composes for the gate's buckets and for the EVENTS consumer, not taken from a composed file.
+//
 // The lists are what mesh-controller's broker.PermissionsFor answers (internal/broker/nats.go:577) for a module principal
 // with the manifest of module.json: consumes the three gitea.pull events, invokes gitea.gitea_open_pulls,
 // gitea.gitea_commit_trees and gitea.gitea_get_file, state heads, and the stream forge; `**` is written `>` on the wire (modstreams.go
@@ -104,6 +108,14 @@ accounts {
            "mesh.mod.gitea.tool.gitea_add_comment"
            "mesh.mod.gitea.tool.gitea_add_comment.*"
            "mesh.stream.mesh-delivery-forge-bridge.event.forge.>"
+            "$JS.ACK.CHECKS.n1_mesh-delivery-forge-bridge.>"
+            "$JS.API.CONSUMER.INFO.CHECKS.n1_mesh-delivery-forge-bridge"
+            "$JS.API.CONSUMER.MSG.NEXT.CHECKS.n1_mesh-delivery-forge-bridge"
+            "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-checker_owed.>"
+            "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-checker_owed.>"
+            "$JS.API.DIRECT.GET.KV_mesh-delivery-checker_owed.>"
+            "$JS.API.STREAM.INFO.KV_mesh-delivery-checker_owed"
+            "$JS.FC.KV_mesh-delivery-checker_owed.>"
          ] }
          subscribe { allow: [
            "_INBOX.n1.mesh-delivery-forge-bridge.>"
@@ -205,6 +217,15 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context,
 	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.>"}}); err != nil {
 		t.Fatal(err)
 	}
+	// The checker's stream checks and the bridge's durable consumer on its owed.written event; the checker's owed bucket is made
+	// only by the test that needs it, as when the checker is assigned after the bridge started.
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CHECKS", Subjects: []string{"mesh.stream.mesh-delivery-checker.event.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.CreateOrUpdateConsumer(ctx, "CHECKS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-forge-bridge", AckPolicy: jetstream.AckExplicitPolicy,
+		FilterSubjects: []string{"mesh.stream.mesh-delivery-checker.event.owed.written"}}); err != nil {
+		t.Fatal(err)
+	}
 	// The bridge's state bucket, made by the controller from the manifest's state entry before the process starts.
 	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: headsBucket, History: 1}); err != nil {
 		t.Fatal(err)
@@ -706,3 +727,65 @@ func forgeMerged(t *testing.T, b *bed) []string {
 }

 func sha256Sum(b []byte) []byte { s := sha256.Sum256(b); return s[:] }
+
+// TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey: the checker's owed bucket is made after the bridge started;
+// the checker writes the status mesh/check it owes on a head (the SDK's CheckerOwedKey and EncodeCheckerOwed, which the checker
+// itself writes with) and says so on its stream; the bridge sets the status on the forge within 10 s, with the context
+// mesh/check, the state and the line; the same revision is not set again, a new revision is (ADR 0484 clause 2).
+func TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T) { // 1683/AC5, ADR 0484 §2
+	b := startBed(t, `{"result":{"pulls":[]}}`, nil)
+	calls := make(chan string, 8)
+	sub, err := b.ctl.Subscribe("mesh.mod.gitea.tool.gitea_commit_status", func(m *nats.Msg) {
+		calls <- string(m.Data)
+		_ = m.Respond([]byte(`{"result":{},"node":"n0"}`))
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	if err := b.ctl.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	kv, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: delivery.CheckerOwedBucket, History: delivery.CheckerOwedHistory})
+	if err != nil {
+		t.Fatal(err)
+	}
+	key, err := delivery.CheckerOwedKey("novox/mesh-delivery", 12, head)
+	if err != nil {
+		t.Fatal(err)
+	}
+	put := func(state, text string) {
+		v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state, Text: text, Target: "https://view.example/checks/g.0123456789ab"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := kv.Put(b.ctx, key, v); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := b.js.Publish(b.ctx, owedWrittenSubject, []byte(`{"key":"`+key+`"}`)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	next := func() string {
+		select {
+		case got := <-calls:
+			return got
+		case <-time.After(10 * time.Second):
+			t.Fatal("the status was not set within 10 s")
+		}
+		return ""
+	}
+	put("pending", "checking 0123456789ab on aaaaaaaaaaaa")
+	if got := next(); !strings.Contains(got, `"context":"mesh/check"`) || !strings.Contains(got, `"state":"pending"`) || !strings.Contains(got, `"sha":"`+head+`"`) || !strings.Contains(got, `"repo":"mesh-delivery"`) {
+		t.Fatalf("first call = %s", got)
+	}
+	put("success", "pass 0123456789ab on aaaaaaaaaaaa | unit 2 run, 3 kept")
+	if got := next(); !strings.Contains(got, `"state":"success"`) || !strings.Contains(got, "unit 2 run, 3 kept") {
+		t.Fatalf("second call = %s", got)
+	}
+	select {
+	case extra := <-calls:
+		t.Errorf("a third call was made: %s", extra)
+	case <-time.After(500 * time.Millisecond):
+	}
+}
@@ -0,0 +1,82 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"strings"
+	"sync"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// checksStream is the stream the checker carries its events on, as the bus names it (declared by the checker as the stream
+// checks, heard by this module through consumes; novox/hq ADR 0484 clause 2).
+const checksStream = "CHECKS"
+
+// owedWrittenSubject is the subject of the checker's owed.written event.
+const owedWrittenSubject = "mesh.stream." + delivery.CheckerModule + ".event.owed.written"
+
+// lateWatch watches a bucket that may be made after the bridge started: the checker's owed bucket. The bridge holds no timer
+// (ADR 0437 §8), so it watches the bucket where it exists at start, and otherwise when the checker says it wrote a status
+// (owed.written, handled by Written).
+type lateWatch struct {
+	ctx context.Context
+	js  interface {
+		KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error)
+	}
+	bucket  string
+	react   func(ctx context.Context, key string, rev uint64, value []byte)
+	logger  *log.Logger
+	errs    chan error
+	mu      sync.Mutex
+	watched bool
+}
+
+// ensure starts the watch if the bucket exists and is not watched.
+func (w *lateWatch) ensure() error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if w.watched {
+		return nil
+	}
+	kv, err := w.js.KeyValue(w.ctx, w.bucket)
+	switch {
+	case errors.Is(err, jetstream.ErrBucketNotFound):
+		w.logger.Printf("the bucket %s does not exist yet: it is watched when the checker says it wrote a status", w.bucket)
+		return nil
+	case err != nil:
+		return fmt.Errorf("the bucket %s: %w", w.bucket, err)
+	}
+	w.watched = true
+	go func() {
+		w.errs <- watch(w.ctx, kv, func(key string, rev uint64, value []byte) {
+			hctx, cancel := context.WithTimeout(w.ctx, syncDeadline)
+			defer cancel()
+			w.react(hctx, key, rev, value)
+		})
+	}()
+	return nil
+}
+
+// Written takes one owed.written event of the checker in: its bucket is watched if it is not. A subject that is not the
+// checker's is terminated; a failure stays unacknowledged, and the server delivers it again.
+func (w *lateWatch) Written(m interface {
+	Subject() string
+	Ack() error
+	Term() error
+}) {
+	if !strings.HasPrefix(m.Subject(), owedWrittenSubject) {
+		w.logger.Printf("%s: not the checker's owed.written, dropped", m.Subject())
+		_ = m.Term()
+		return
+	}
+	if err := w.ensure(); err != nil {
+		w.logger.Printf("%s: left for redelivery: %v", m.Subject(), err)
+		return
+	}
+	_ = m.Ack()
+}
@@ -202,7 +202,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return fmt.Errorf("the bucket %s: %w", performedBucket, err)
 	}
 	perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger}
-	watching := make(chan error, 2)
+	watching := make(chan error, 3)
 	for _, w := range []struct {
 		bucket string
 		react  func(ctx context.Context, key string, rev uint64, value []byte)
@@ -225,6 +225,24 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			})
 		}()
 	}
+	// The checker's statuses (mesh/check, unrequired until release R4f): its bucket is made when the checker is assigned, which
+	// may be after the bridge started, and the bridge holds no timer; the checker says on its stream each time it wrote a
+	// status, and the bucket is then watched (late.go).
+	late := &lateWatch{ctx: ctx, js: js, bucket: delivery.CheckerOwedBucket, react: perf.checkOwed, logger: logger, errs: watching}
+	if err := late.ensure(); err != nil {
+		return err
+	}
+	switch ccons, err := js.Consumer(ctx, checksStream, cred.Node+"_"+cred.Module); {
+	case err != nil:
+		logger.Printf("the consumer %s_%s on %s does not exist yet, so the checker's statuses are not heard until the bridge starts again: %v", cred.Node, cred.Module, checksStream, err)
+	default:
+		cc, err := ccons.Consume(func(m jetstream.Msg) { late.Written(m) },
+			jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
+		if err != nil {
+			return err
+		}
+		defer cc.Stop()
+	}
 	if err := said(ctx); err != nil {
 		return err
 	}