Perform the gate's owed acts once per revision from the bridge, a refusal written on FORGE (issue 1341)

2026-10-20 00:16:43 +00:00
parent a0883a174b
commit eeb5561a23
4 changed files with 367 additions and 2 deletions
@@ -0,0 +1,137 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"strconv"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-forge-bridge/internal/bridge"
+)
+
+// performedBucket is the module's state bucket performed, `<module>_<name>` as heads is: for each key of the gate's
+// buckets the bridge has acted on, the revision it acted at (manifest state entry performed, novox/hq ADR 0437 §8).
+// Its keys are `owed.<owed key>` and `merges.<merges key>`; its value is the revision in decimal.
+const performedBucket = delivery.ForgeBridgeModule + "_performed"
+
+// kvRevisions is the bucket performed as the performer's revisions.
+type kvRevisions struct{ kv jetstream.KeyValue }
+
+func (s kvRevisions) Get(ctx context.Context, key string) (uint64, bool, error) {
+	e, err := s.kv.Get(ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return 0, false, nil
+	}
+	if err != nil {
+		return 0, false, err
+	}
+	rev, err := strconv.ParseUint(string(e.Value()), 10, 64)
+	if err != nil {
+		return 0, false, fmt.Errorf("%s holds %q, which is no revision: %w", key, e.Value(), err)
+	}
+	return rev, true, nil
+}
+
+func (s kvRevisions) Put(ctx context.Context, key string, rev uint64) error {
+	_, err := s.kv.PutString(ctx, key, strconv.FormatUint(rev, 10))
+	return err
+}
+
+// revisions is what the performer keeps of the revisions it acted at.
+type revisions interface {
+	Get(ctx context.Context, key string) (rev uint64, found bool, err error)
+	Put(ctx context.Context, key string, rev uint64) error
+}
+
+// performer does on the forge what the gate's keys ask (novox/hq ADR 0437 §8): one act for each revision of an owed
+// key. A refusal is written on FORGE and the act is not tried again until
+// the key has a new revision; the performer holds no timer and serves no request.
+type performer struct {
+	call   caller
+	pub    publisher
+	done   revisions
+	logger *log.Logger
+}
+
+// seen says whether the revision of the key was acted on already. A revision that cannot be read is not seen, so the
+// act is done: a status set twice is the same status.
+func (p *performer) seen(ctx context.Context, doneKey string, rev uint64) bool {
+	kept, found, err := p.done.Get(ctx, doneKey)
+	if err != nil {
+		p.logger.Printf("%s: the revision acted at could not be read, so the act is done again: %v", doneKey, err)
+		return false
+	}
+	return found && kept >= rev
+}
+
+func (p *performer) keep(ctx context.Context, doneKey string, rev uint64) {
+	if err := p.done.Put(ctx, doneKey, rev); err != nil {
+		p.logger.Printf("%s: revision %d was acted on but could not be kept, so a restart acts on it again: %v", doneKey, rev, err)
+	}
+}
+
+// refuse writes the refused message of an act on FORGE; it reports whether it was written. The message id makes a
+// second write of one revision's refusal the same message to the stream.
+func (p *performer) refuse(ctx context.Context, doneKey string, rev uint64, a bridge.Act, words string) bool {
+	msg, err := bridge.RefusedMessage(a, words)
+	if err != nil {
+		p.logger.Printf("%s: the forge refused %s (%s) and the refusal cannot be said on FORGE: %v", doneKey, a.Name, words, err)
+		return false
+	}
+	id := fmt.Sprintf("refused.%s.%d", doneKey, rev)
+	if _, err := p.pub.Publish(ctx, forgeSubject(msg.Event), msg.Data, jetstream.WithMsgID(id)); err != nil {
+		p.logger.Printf("%s: the forge refused %s (%s) and FORGE could not be written: %v", doneKey, a.Name, words, err)
+		return false
+	}
+	return true
+}
+
+// owed performs the act an owed key asks at revision rev, once.
+func (p *performer) owed(ctx context.Context, key string, rev uint64, value []byte) {
+	doneKey := "owed." + key
+	if p.seen(ctx, doneKey, rev) {
+		return
+	}
+	a, err := bridge.OwedAct(key, value)
+	if err != nil {
+		p.logger.Printf("%s at revision %d is not an act the bridge can perform, and is left: %v", key, rev, err)
+		p.keep(ctx, doneKey, rev)
+		return
+	}
+	if _, err := p.call(ctx, a.Call.Tool, a.Call.Args); err != nil {
+		p.logger.Printf("%s at revision %d: %s refused: %v", key, rev, a.Call.Tool, err)
+		if !p.refuse(ctx, doneKey, rev, a, err.Error()) {
+			return
+		}
+	}
+	p.keep(ctx, doneKey, rev)
+}
+
+// watch hands each key of a bucket, with the revision it has now and then each new one, to react until ctx ends or the
+// watch fails. A deleted key is not an act.
+func watch(ctx context.Context, kv jetstream.KeyValue, react func(key string, rev uint64, value []byte)) error {
+	w, err := kv.Watch(ctx, ">")
+	if err != nil {
+		return err
+	}
+	defer w.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case e, ok := <-w.Updates():
+			if !ok {
+				return errors.New("the watch ended")
+			}
+			if e == nil || e.Operation() != jetstream.KeyValuePut {
+				continue
+			}
+			react(e.Key(), e.Revision(), e.Value())
+		}
+	}
+}
@@ -0,0 +1,147 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"log"
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+const (
+	headA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	headB = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+)
+
+type memRevisions map[string]uint64
+
+func (m memRevisions) Get(_ context.Context, key string) (uint64, bool, error) {
+	r, ok := m[key]
+	return r, ok, nil
+}
+func (m memRevisions) Put(_ context.Context, key string, rev uint64) error { m[key] = rev; return nil }
+
+type tcall struct {
+	key  string
+	body string
+}
+
+// forgeFake records the calls and answers each from refuse, a map from a tool to its refusal.
+type forgeFake struct {
+	calls  []tcall
+	refuse map[string]string
+}
+
+func (f *forgeFake) call(_ context.Context, key string, body any) (json.RawMessage, error) {
+	b, _ := json.Marshal(body)
+	f.calls = append(f.calls, tcall{key, string(b)})
+	if w, ok := f.refuse[key]; ok {
+		return nil, errors.New(w)
+	}
+	return json.RawMessage(`{}`), nil
+}
+
+func newPerformer(f *forgeFake) (*performer, *fakePub, memRevisions, *bytes.Buffer) {
+	pub, done, logs := &fakePub{}, memRevisions{}, &bytes.Buffer{}
+	return &performer{call: f.call, pub: pub, done: done, logger: log.New(logs, "", 0)}, pub, done, logs
+}
+
+// The values are the SDK's own encoding (mesh-sdk go/delivery EncodeOwed, EncodeMerge), the keys its OwedKey.
+func owedKV(t *testing.T, repository string, number int, act string, o delivery.Owed) (string, []byte) {
+	t.Helper()
+	key, err := delivery.OwedKey(repository, number, headA, act)
+	if err != nil {
+		t.Fatal(err)
+	}
+	v, err := delivery.EncodeOwed(o)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return key, v
+}
+
+func TestAnOwedStatusIsSetOnceForARevision(t *testing.T) { // 1341/AC1
+	f := &forgeFake{}
+	p, _, _, _ := newPerformer(f)
+	key, v := owedKV(t, "novox/mesh-delivery", 12, "status=mesh/merge-gate",
+		delivery.Owed{Act: delivery.OwedStatus, Context: "mesh/merge-gate", State: "success", Text: "all green", Target: "https://forge.example/p/12"})
+	p.owed(context.Background(), key, 7, v)
+	if len(f.calls) != 1 || f.calls[0].key != "gitea.gitea_commit_status" ||
+		f.calls[0].body != `{"context":"mesh/merge-gate","description":"all green","owner":"novox","repo":"mesh-delivery","sha":"`+headA+`","state":"success","target_url":"https://forge.example/p/12"}` {
+		t.Fatalf("calls = %+v", f.calls)
+	}
+	p.owed(context.Background(), key, 7, v) // the same revision seen again (a restart's replay)
+	if len(f.calls) != 1 {
+		t.Errorf("the same revision called the tool again: %+v", f.calls)
+	}
+	p.owed(context.Background(), key, 8, v) // a new revision of the key is a new act
+	if len(f.calls) != 2 {
+		t.Errorf("a new revision was not performed: %+v", f.calls)
+	}
+}
+
+func TestViewNoteAndCommentEachCallTheirToolOnce(t *testing.T) { // 1341/AC2
+	f := &forgeFake{}
+	p, _, _, _ := newPerformer(f)
+	for _, c := range []struct {
+		act string
+		o   delivery.Owed
+	}{
+		{"view", delivery.Owed{Act: delivery.OwedView, Text: "the view"}},
+		{"note", delivery.Owed{Act: delivery.OwedNote, Text: "a line", Ref: "mesh-plan"}},
+		{"comment.1", delivery.Owed{Act: delivery.OwedComment, Text: "hello"}},
+	} {
+		key, v := owedKV(t, "novox/mesh-delivery", 12, c.act, c.o)
+		p.owed(context.Background(), key, 1, v)
+		p.owed(context.Background(), key, 1, v)
+	}
+	var tools []string
+	for _, c := range f.calls {
+		tools = append(tools, c.key)
+	}
+	if got := strings.Join(tools, " "); got != "gitea.gitea_delivery_view gitea.gitea_note_append gitea.gitea_add_comment" {
+		t.Errorf("calls = %s", got)
+	}
+}
+
+func TestARefusedActIsWrittenOnForgeOnceAndNotRetried(t *testing.T) { // 1341/AC3
+	f := &forgeFake{refuse: map[string]string{"gitea.gitea_add_comment": "the pull request is locked"}}
+	p, pub, _, _ := newPerformer(f)
+	key, v := owedKV(t, "novox/mesh-delivery", 12, "comment.1", delivery.Owed{Act: delivery.OwedComment, Text: "hello"})
+	p.owed(context.Background(), key, 3, v)
+	p.owed(context.Background(), key, 3, v)
+	if len(f.calls) != 1 {
+		t.Errorf("the refused act was tried %d times at one revision", len(f.calls))
+	}
+	if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.refused" {
+		t.Fatalf("published on %v", pub.subjects)
+	}
+	m, err := delivery.DecodeForge(pub.data[0])
+	if err != nil || m.Kind != delivery.ForgeRefused || m.Act != "comment.1" || m.Number != 12 || m.Head != headA || m.Words != "the pull request is locked" {
+		t.Errorf("message %+v err %v", m, err)
+	}
+	f.refuse = nil
+	p.owed(context.Background(), key, 4, v) // only a new revision tries again
+	if len(f.calls) != 2 {
+		t.Errorf("a new revision was not tried: %d calls", len(f.calls))
+	}
+}
+
+func TestAnOwedValueThatCannotBeReadIsLoggedAndNotRetried(t *testing.T) {
+	f := &forgeFake{}
+	p, pub, done, logs := newPerformer(f)
+	p.owed(context.Background(), "not-a-key", 2, []byte(`{}`))
+	if len(f.calls) != 0 || len(pub.subjects) != 0 {
+		t.Errorf("calls %v published %v", f.calls, pub.subjects)
+	}
+	if !strings.Contains(logs.String(), "not-a-key") {
+		t.Errorf("log %q does not name the key", logs.String())
+	}
+	if rev, ok := done["owed.not-a-key"]; !ok || rev != 2 {
+		t.Errorf("revision kept = %v %v", rev, ok)
+	}
+}
@@ -25,6 +25,9 @@ import (
 // the controller's user, and the bridge's own principal <node>.<module> with the permissions the controller composes for
 // it, and no others, so a permission the bridge lacks fails a test here and not on the mesh (issue 1279).
 //
+// 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 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 and
 // gitea.gitea_commit_trees, state heads, and the stream forge; `**` is written `>` on the wire (modstreams.go
@@ -58,6 +61,44 @@ accounts {
            "mesh.mod.gitea.tool.gitea_commit_trees.*"
            "mesh.mod.gitea.tool.gitea_open_pulls"
            "mesh.mod.gitea.tool.gitea_open_pulls.*"
+            "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-forge-bridge_performed.>"
+            "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-forge-bridge_performed.>"
+            "$JS.API.DIRECT.GET.KV_mesh-delivery-forge-bridge_performed.>"
+            "$JS.API.STREAM.INFO.KV_mesh-delivery-forge-bridge_performed"
+            "$JS.FC.KV_mesh-delivery-forge-bridge_performed.>"
+            "$KV.mesh-delivery-forge-bridge_performed.>"
+            "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-gate_merges.>"
+            "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-gate_merges.>"
+            "$JS.API.DIRECT.GET.KV_mesh-delivery-gate_merges.>"
+            "$JS.API.STREAM.INFO.KV_mesh-delivery-gate_merges"
+            "$JS.FC.KV_mesh-delivery-gate_merges.>"
+            "$KV.mesh-delivery-gate_merges.>"
+            "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-gate_owed.>"
+            "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-gate_owed.>"
+            "$JS.API.DIRECT.GET.KV_mesh-delivery-gate_owed.>"
+            "$JS.API.STREAM.INFO.KV_mesh-delivery-gate_owed"
+            "$JS.FC.KV_mesh-delivery-gate_owed.>"
+            "$KV.mesh-delivery-gate_owed.>"
+            "mesh.mod.gitea.call.gitea_merge_pull_request.*.n1~mesh-delivery-forge-bridge"
+            "mesh.mod.gitea.call.gitea_merge_pull_request.n1~mesh-delivery-forge-bridge"
+            "mesh.mod.gitea.tool.gitea_merge_pull_request"
+            "mesh.mod.gitea.tool.gitea_merge_pull_request.*"
+            "mesh.mod.gitea.call.gitea_commit_status.*.n1~mesh-delivery-forge-bridge"
+            "mesh.mod.gitea.call.gitea_commit_status.n1~mesh-delivery-forge-bridge"
+            "mesh.mod.gitea.tool.gitea_commit_status"
+            "mesh.mod.gitea.tool.gitea_commit_status.*"
+            "mesh.mod.gitea.call.gitea_delivery_view.*.n1~mesh-delivery-forge-bridge"
+            "mesh.mod.gitea.call.gitea_delivery_view.n1~mesh-delivery-forge-bridge"
+            "mesh.mod.gitea.tool.gitea_delivery_view"
+            "mesh.mod.gitea.tool.gitea_delivery_view.*"
+            "mesh.mod.gitea.call.gitea_note_append.*.n1~mesh-delivery-forge-bridge"
+            "mesh.mod.gitea.call.gitea_note_append.n1~mesh-delivery-forge-bridge"
+            "mesh.mod.gitea.tool.gitea_note_append"
+            "mesh.mod.gitea.tool.gitea_note_append.*"
+            "mesh.mod.gitea.call.gitea_add_comment.*.n1~mesh-delivery-forge-bridge"
+            "mesh.mod.gitea.call.gitea_add_comment.n1~mesh-delivery-forge-bridge"
+            "mesh.mod.gitea.tool.gitea_add_comment"
+            "mesh.mod.gitea.tool.gitea_add_comment.*"
            "mesh.stream.mesh-delivery-forge-bridge.event.forge.>"
          ] }
          subscribe { allow: [
@@ -149,6 +190,13 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context,
 	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: headsBucket, History: 1}); err != nil {
 		t.Fatal(err)
 	}
+	// Its other state bucket, and the gate's buckets the bridge reads, made by the controller from the gate's manifest
+	// (mesh-sdk go/delivery MergesBucket, OwedBucket with their histories).
+	for name, history := range map[string]uint8{performedBucket: 1, delivery.MergesBucket: delivery.MergesHistory, delivery.OwedBucket: delivery.OwedHistory} {
+		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
+			t.Fatal(err)
+		}
+	}
 	if prepare != nil {
 		prepare(ctx, js)
 	}
@@ -196,6 +196,35 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// Everything written on FORGE, an event's and a sync's, keeps the head in the bucket.
 	rec := &recorder{pub: js, heads: heads, logger: logger}
 	said := func(ctx context.Context) error { return syncOpen(ctx, call, rec, heads, &turn, logger) }
+	// What the gate asks of the forge (ADR 0437 §8): its owed acts and its merges, each performed once per revision.
+	performedKV, err := js.KeyValue(ctx, performedBucket)
+	if err != nil {
+		return fmt.Errorf("the bucket %s: %w", performedBucket, err)
+	}
+	perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger}
+	watching := make(chan error, 1)
+	for _, w := range []struct {
+		bucket string
+		react  func(ctx context.Context, key string, rev uint64, value []byte)
+	}{{delivery.OwedBucket, perf.owed}} {
+		kv, err := js.KeyValue(ctx, w.bucket)
+		if errors.Is(err, jetstream.ErrBucketNotFound) {
+			// The gate's buckets come with the gate; the bridge does not wait for them (no timer), and is started again
+			// by its delivery when the gate arrives.
+			logger.Printf("the bucket %s does not exist yet: nothing of it is performed until the bridge starts again", w.bucket)
+			continue
+		}
+		if err != nil {
+			return fmt.Errorf("the bucket %s: %w", w.bucket, err)
+		}
+		go func() {
+			watching <- watch(ctx, kv, func(key string, rev uint64, value []byte) {
+				hctx, cancel := context.WithTimeout(ctx, syncDeadline)
+				defer cancel()
+				w.react(hctx, key, rev, value)
+			})
+		}()
+	}
 	if err := said(ctx); err != nil {
 		return err
 	}
@@ -217,6 +246,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return err
 	}
 	defer cc.Stop()
-	<-ctx.Done()
-	return nil
+	select {
+	case <-ctx.Done():
+		return nil
+	case err := <-watching:
+		return fmt.Errorf("watching the gate's keys: %w", err)
+	}
 }