From eeb5561a23155a6d908d53cd78e6444c5904ff52 Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 00:16:43 +0000 Subject: [PATCH] Perform the gate's owed acts once per revision from the bridge, a refusal written on FORGE (issue 1341) --- .../cmd/mesh-delivery-forge-bridge/gate.go | 137 ++++++++++++++++ .../mesh-delivery-forge-bridge/gate_test.go | 147 ++++++++++++++++++ .../integration_test.go | 48 ++++++ .../cmd/mesh-delivery-forge-bridge/main.go | 37 ++++- 4 files changed, 367 insertions(+), 2 deletions(-) create mode 100644 modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate.go create mode 100644 modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate.go new file mode 100644 index 0000000..9de2289 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate.go @@ -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, `_` 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.` and `merges.`; 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()) + } + } +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go new file mode 100644 index 0000000..1d880bf --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go index 6769ff0..4dc5e66 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go @@ -25,6 +25,9 @@ import ( // the controller's user, and the bridge's own principal . 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) } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go index c067c25..68e03c7 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go @@ -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) + } }