From efca168c552251c99974f825dbc76908f60e67aa Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 21:40:09 +0200 Subject: [PATCH] Stop the forge bridge performing mesh/check: the forge module writes it (R2-1) R4e round 2 finding R2-1, ADR 0521: nothing in the delivery plan orders the forge module before the bridge, and the forge module's durable consumer keeps its deliver policy when its filters change, so a check.status emitted before it consumed the event was lost. The forge module now watches the checker's owed state and writes mesh/check itself, once per revision. The bridge emits no check.status, declares no emits, and neither reads the checker's owed nor hears its stream; OwedAct still refuses mesh/check. The integration test shows the bridge calls no tool and publishes nothing of its own when the checker owes mesh/check. --- .../cmd/mesh-delivery-forge-bridge/gate.go | 54 ----- .../mesh-delivery-forge-bridge/gate_test.go | 104 --------- .../integration_test.go | 221 +++--------------- .../cmd/mesh-delivery-forge-bridge/late.go | 82 ------- .../cmd/mesh-delivery-forge-bridge/main.go | 20 +- .../internal/bridge/acts.go | 56 +---- .../internal/bridge/acts_test.go | 38 +-- .../manifest_test.go | 14 +- .../mesh-delivery-forge-bridge/module.json | 7 +- 9 files changed, 43 insertions(+), 553 deletions(-) delete mode 100644 modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/late.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 index dedcb0a..78ac361 100644 --- 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 @@ -6,9 +6,7 @@ import ( "fmt" "log" "strconv" - "time" - "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -58,20 +56,8 @@ type performer struct { pub publisher done revisions logger *log.Logger - // emit publishes the bridge's own events (check.status), and node is the machine said in their headers. - emit emitter - node string } -// emitter publishes a message with its headers: the bus's JetStream. -type emitter interface { - PublishMsg(ctx context.Context, m *nats.Msg, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error) -} - -// checkStatusSubject is where the bridge emits check.status: under its own name, which the bus lets its account alone -// publish (mesh-control internal/broker/nats.go, a module's own namespace). -const checkStatusSubject = "mesh.mod." + delivery.ForgeBridgeModule + ".event." + bridge.CheckStatusEvent - // 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 { @@ -110,46 +96,6 @@ func (p *performer) owed(ctx context.Context, key string, rev uint64, value []by 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 - } - // mesh/check is emitted on the bridge's own subject, never written through a tool (R4e finding K1): the forge module - // consumes check.status and writes it. The event's id is the bus's dedup id, so a second emit of a revision is one. - id, body, err := bridge.CheckStatusMessage(key, rev, value) - if 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 p.emit == nil { - p.logger.Printf("%s at revision %d: nothing to emit %s with, so it is not kept and a restart emits it", key, rev, bridge.CheckStatusEvent) - return - } - m := nats.NewMsg(checkStatusSubject) - m.Header.Set("x-event-id", id) - m.Header.Set("Nats-Msg-Id", id) - m.Header.Set("x-source", delivery.ForgeBridgeModule) - m.Header.Set("x-node", p.node) - m.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) - m.Header.Set("content-type", "application/json") - m.Data = body - if _, err := p.emit.PublishMsg(ctx, m); err != nil { - p.logger.Printf("%s at revision %d: %s could not be emitted, and is emitted again when the key is seen again: %v", key, rev, bridge.CheckStatusEvent, err) - return - } - p.keep(ctx, doneKey, rev) -} - // 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) { 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 index 0c13452..608685e 100644 --- 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 @@ -9,9 +9,6 @@ import ( "strings" "testing" - "github.com/nats-io/nats.go" - "github.com/nats-io/nats.go/jetstream" - "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -211,104 +208,3 @@ 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 -} - -// fakeEmit keeps what the performer emits on its own subjects. -type fakeEmit struct { - msgs []*nats.Msg - err error -} - -func (e *fakeEmit) PublishMsg(_ context.Context, m *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) { - if e.err != nil { - return nil, e.err - } - e.msgs = append(e.msgs, m) - return &jetstream.PubAck{}, nil -} - -// R4e finding K1 (issue 1301 S4): the checker's status is emitted as check.status on the bridge's own subject, never a -// tool call, once per revision of its owed key (1683/AC5, ADR 0484 clause 2). -func TestTheCheckersStatusIsEmittedOncePerRevisionAndAKeyOfTheGateIsNotItsKey(t *testing.T) { // K1, 1683/AC5 - f := &forgeFake{} - p, _, done, _ := newPerformer(f) - em := &fakeEmit{} - p.emit, p.node = em, "n1" - 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) != 0 { - t.Fatalf("a tool was called for mesh/check: %+v", f.calls) - } - if len(em.msgs) != 1 { - t.Fatalf("emitted %d, want 1", len(em.msgs)) - } - m := em.msgs[0] - if m.Subject != "mesh.mod.mesh-delivery-forge-bridge.event.check.status" { - t.Errorf("subject = %q", m.Subject) - } - if m.Header.Get("x-source") != "mesh-delivery-forge-bridge" || m.Header.Get("x-node") != "n1" || m.Header.Get("x-time") == "" || - m.Header.Get("content-type") != "application/json" || m.Header.Get("x-event-id") != "check."+key+".3" || m.Header.Get("Nats-Msg-Id") != "check."+key+".3" { - t.Errorf("headers = %v", m.Header) - } - if want := `{"owner":"novox","repo":"mesh-delivery","sha":"` + headA + `","state":"success","description":"pass aaaaaaaaaaaa on bbbbbbbbbbbb","target_url":"https://view.example/checks/g.0123456789ab","key":"` + key + `","revision":3}`; string(m.Data) != want { - t.Errorf("body = %s", m.Data) - } - p.checkOwed(context.Background(), key, 4, v) - if len(em.msgs) != 2 { - t.Errorf("a new revision was not emitted: %d", len(em.msgs)) - } - 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 TestACheckStatusTheBusDidNotTakeIsNotKeptAsDone(t *testing.T) { // K1 - p, _, done, logs := newPerformer(&forgeFake{}) - em := &fakeEmit{err: errors.New("nats: timeout")} - p.emit = em - key, v := checkerOwed(t, 12, "pending", "checking") - p.checkOwed(context.Background(), key, 5, v) - if _, ok := done["check."+key]; ok || !strings.Contains(logs.String(), "nats: timeout") { - t.Fatalf("kept %v, log %q: a status the bus did not take is said and not kept", done, logs.String()) - } - em.err = nil - p.checkOwed(context.Background(), key, 5, v) - if len(em.msgs) != 1 { - t.Errorf("the same revision was not emitted again after the failure: %d", len(em.msgs)) - } -} - -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) - } -} 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 8a5ca63..05497d4 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 @@ -30,15 +30,8 @@ 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 user n1.node-tools is the machine's tool runner, whose account every agent session on the machine shares, carrying the -// forge module (mesh-control internal/broker users.go KindNodeTools): it invokes every tool (node-tools' invokes "*"), -// publishes gitea's own events and hears gitea's consumer on EVENTS, n1_node-tools, which binds the bridge's check.status -// (gitea's consumes). Written by analogy with what the controller composes, as the checker's entries are; it is never -// granted the bridge's namespace, which is the point of R4e finding K1. +// The bridge neither reads the checker's owed nor hears its stream (ADR 0521, R4e finding R2-1): the forge module writes +// mesh/check from it. The stream checks is made here so a test can say the checker wrote a status. // // 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, @@ -114,15 +107,6 @@ accounts { "mesh.mod.gitea.tool.gitea_add_comment" "mesh.mod.gitea.tool.gitea_add_comment.*" "mesh.stream.mesh-delivery-forge-bridge.event.forge.>" - "mesh.mod.mesh-delivery-forge-bridge.event.check.status" - "$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.>" @@ -136,24 +120,6 @@ accounts { allow_responses: true } } - { user: "n1.node-tools", password: "pw" - permissions { - publish { allow: [ - "mesh.mod.*.tool.>" - "mesh.mod.*.call.>" - "mesh.mod.gitea.event.>" - "$JS.ACK.EVENTS.n1_node-tools.>" - "$JS.API.CONSUMER.INFO.EVENTS.n1_node-tools" - "$JS.API.CONSUMER.MSG.NEXT.EVENTS.n1_node-tools" - ] } - subscribe { allow: [ - "_INBOX.n1.node-tools.>" - "mesh.mod.gitea.tool.>" - "mesh.mod.mesh-delivery-forge-bridge.event.check.status" - ] } - allow_responses: true - } - } ] } } @@ -242,15 +208,10 @@ 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. + // The checker's stream checks, which the bridge no longer hears (ADR 0521): a test says on it that the checker wrote a status. 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) @@ -753,45 +714,13 @@ func forgeMerged(t *testing.T, b *bed) []string { func sha256Sum(b []byte) []byte { s := sha256.Sum256(b); return s[:] } -// checkStatusAt is the subject the bridge emits check.status on, and the one gitea's consumer binds. -const checkStatusAt = "mesh.mod.mesh-delivery-forge-bridge.event.check.status" - -// gitesConsumer is the durable consumer of the tool runner carrying gitea on EVENTS, as the controller makes it from gitea's -// consumes (nats.go consumerDurable _), filtered on the bridge's check.status. -func giteaConsumer(ctx context.Context, t *testing.T, js jetstream.JetStream) { - t.Helper() - if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_node-tools", AckPolicy: jetstream.AckExplicitPolicy, - FilterSubjects: []string{checkStatusAt}}); err != nil { - t.Fatal(err) - } -} - -// asRunner connects as the machine's tool runner, the account agent sessions share. -func asRunner(t *testing.T, b *bed) (*nats.Conn, jetstream.JetStream) { - t.Helper() - nc, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo("n1.node-tools", "pw"), nats.CustomInboxPrefix("_INBOX.n1.node-tools")) - if err != nil { - t.Fatal(err) - } - t.Cleanup(nc.Close) - js, err := jetstream.New(nc, jetstream.WithDefaultTimeout(2*time.Second)) - if err != nil { - t.Fatal(err) - } - return nc, js -} - -// TestBridgeEmitsTheCheckersStatusOncePerRevisionOfItsOwedKey: 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 emits check.status on its own subject within 10 s, with the -// state, the line, the key and the revision; the same revision is not emitted again, a new revision is (ADR 0484 clause 2). -// No tool is called for it (R4e finding K1). Both statuses are emitted before gitea's runner fetches anything, and its -// durable consumer hands both over afterwards: a check.status emitted while gitea's process is not running is applied -// when it runs. -func TestBridgeEmitsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T) { // K1, 1683/AC5, ADR 0484 §2 - b := startBed(t, `{"result":{"pulls":[]}}`, func(ctx context.Context, js jetstream.JetStream) { giteaConsumer(ctx, t, js) }) +// TestBridgePerformsNoMeshCheckAct: the checker owes mesh/check on a head and says so on its stream; the bridge calls no +// tool with mesh/check and publishes nothing of its own on EVENTS for it. The forge module writes mesh/check from the +// checker's owed state itself (ADR 0521; R4e round 2 finding R2-1). +func TestBridgePerformsNoMeshCheckAct(t *testing.T) { // R2-1, ADR 0521 + b := startBed(t, `{"result":{"pulls":[]}}`, nil) calls := make(chan string, 8) - // Every tool call is watched, and none answered here: the bed's stand-ins answer the ones the bridge asks. + // Every forge call is watched, and none answered here: the bed's stand-ins answer the ones the bridge asks. sub, err := b.ctl.Subscribe("mesh.mod.gitea.>", func(m *nats.Msg) { if strings.Contains(string(m.Data), "mesh/check") { calls <- m.Subject + " " + string(m.Data) @@ -812,124 +741,26 @@ func TestBridgeEmitsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T) { if err != nil { t.Fatal(err) } + v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", 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, "mesh.stream.mesh-delivery-checker.event.owed.written", []byte(`{"key":"`+key+`"}`)); err != nil { + t.Fatal(err) + } + select { + case c := <-calls: + t.Errorf("a forge tool was called with mesh/check: %s", c) + case <-time.After(2 * time.Second): + } events, err := b.js.Stream(b.ctx, "EVENTS") if err != nil { t.Fatal(err) } - emitted := func() uint64 { - info, err := events.Info(b.ctx, jetstream.WithSubjectFilter(checkStatusAt)) - if err != nil { - t.Fatal(err) - } - return info.State.Msgs - } - put := func(state, text string, want uint64) uint64 { - 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) - } - rev, err := kv.Put(b.ctx, key, v) - if err != nil { - t.Fatal(err) - } - if _, err := b.js.Publish(b.ctx, owedWrittenSubject, []byte(`{"key":"`+key+`"}`)); err != nil { - t.Fatal(err) - } - for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline) && emitted() < want; time.Sleep(50 * time.Millisecond) { - } - if got := emitted(); got != want { - t.Fatalf("EVENTS holds %d check.status within 10 s, want %d", got, want) - } - return rev - } - rev1 := put("pending", "checking 0123456789ab on aaaaaaaaaaaa", 1) - rev2 := put("success", "pass 0123456789ab on aaaaaaaaaaaa | unit 2 run, 3 kept", 2) - time.Sleep(500 * time.Millisecond) - if got := emitted(); got != 2 { - t.Errorf("EVENTS holds %d check.status, want 2: a revision was emitted twice", got) - } - select { - case c := <-calls: - t.Errorf("a tool was called with mesh/check: %s", c) - default: - } - - // gitea's runner starts fetching only now, and is handed both, in order. - _, rjs := asRunner(t, b) - cons, err := rjs.Consumer(b.ctx, "EVENTS", "n1_node-tools") - if err != nil { - t.Fatal(err) - } - batch, err := cons.Fetch(2, jetstream.FetchMaxWait(5*time.Second)) - if err != nil { - t.Fatal(err) - } - var got []string - for m := range batch.Messages() { - var c struct { - Owner, Repo, Sha, State, Description, Key string - Revision uint64 - } - if err := json.Unmarshal(m.Data(), &c); err != nil { - t.Fatal(err) - } - if m.Headers().Get("x-source") != "mesh-delivery-forge-bridge" || m.Headers().Get("x-event-id") != fmt.Sprintf("check.%s.%d", c.Key, c.Revision) { - t.Errorf("headers = %v", m.Headers()) - } - got = append(got, fmt.Sprintf("%s/%s@%s %s %d %v", c.Owner, c.Repo, c.Sha[:8], c.State, c.Revision, c.Key == key)) - _ = m.Ack() - } - want := []string{fmt.Sprintf("novox/mesh-delivery@%s pending %d true", head[:8], rev1), fmt.Sprintf("novox/mesh-delivery@%s success %d true", head[:8], rev2)} - if strings.Join(got, "|") != strings.Join(want, "|") { - t.Errorf("gitea's consumer was handed %v, want %v", got, want) - } -} - -// TestOnlyTheBridgesAccountMayEmitCheckStatus: the machine's tool runner, whose account every agent session there shares and -// which invokes every tool, publishing on the bridge's check.status subject is refused by the bus, and gitea's consumer is -// handed nothing (R4e finding K1). -func TestOnlyTheBridgesAccountMayEmitCheckStatus(t *testing.T) { // K1 - b := startBed(t, `{"result":{"pulls":[]}}`, func(ctx context.Context, js jetstream.JetStream) { giteaConsumer(ctx, t, js) }) - refused := make(chan error, 4) - nc, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo("n1.node-tools", "pw"), nats.CustomInboxPrefix("_INBOX.n1.node-tools"), - nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused <- err })) - if err != nil { - t.Fatal(err) - } - t.Cleanup(nc.Close) - js, err := jetstream.New(nc) - if err != nil { - t.Fatal(err) - } - forged, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-delivery", "sha": head, "state": "success", - "description": "by a session", "key": "x", "revision": 1 << 40}) - m := nats.NewMsg(checkStatusAt) - m.Header.Set("x-source", "mesh-delivery-forge-bridge") - m.Data = forged - ctx, cancel := context.WithTimeout(b.ctx, 2*time.Second) - defer cancel() - if _, err := js.PublishMsg(ctx, m); err == nil { - t.Fatal("the tool runner's account published check.status: any session could set mesh/check") - } - select { - case err := <-refused: - if !strings.Contains(err.Error(), "Permissions Violation for Publish") || !strings.Contains(err.Error(), checkStatusAt) { - t.Errorf("refused with %v", err) - } - case <-time.After(5 * time.Second): - t.Error("the bus said no permissions violation for the runner's publish") - } - // And the ordinary way in, a tool call, has no tool that writes mesh/check to call: the runner may call every tool, so - // the refusal is the bus's, not a manifest's. - info, err := b.js.Stream(b.ctx, "EVENTS") - if err != nil { - t.Fatal(err) - } - si, err := info.Info(b.ctx, jetstream.WithSubjectFilter(checkStatusAt)) - if err != nil { - t.Fatal(err) - } - if si.State.Msgs != 0 { - t.Errorf("EVENTS holds %d check.status after the runner's publish, want 0", si.State.Msgs) + if _, err := events.GetLastMsgForSubject(b.ctx, "mesh.mod.mesh-delivery-forge-bridge.event.>"); err == nil { + t.Error("the bridge published an event of its own for the checker's mesh/check") } } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/late.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/late.go deleted file mode 100644 index c25f87a..0000000 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/late.go +++ /dev/null @@ -1,82 +0,0 @@ -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() -} 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 728789c..a11a3b2 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 @@ -201,7 +201,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return fmt.Errorf("the bucket %s: %w", performedBucket, err) } - perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger, emit: js, node: cred.Node} + perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger} watching := make(chan error, 3) for _, w := range []struct { bucket string @@ -225,24 +225,6 @@ 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 } diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/acts.go b/modules/mesh-delivery-forge-bridge/internal/bridge/acts.go index 4faeed4..f35ce98 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/acts.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/acts.go @@ -1,9 +1,7 @@ package bridge import ( - "encoding/json" "fmt" - "regexp" "slices" "strings" @@ -39,54 +37,6 @@ const ( mergePullTool = "gitea.gitea_merge_pull_request" ) -// CheckStatusEvent is the event the bridge writes mesh/check with (R4e finding K1; issue 1301 S4; ADR 0503 §3): no tool -// writes mesh/check, since a tool is told no caller and node-tools invokes every tool. The bus grants its subject, -// mesh.mod.mesh-delivery-forge-bridge.event.check.status, to this module's own account alone (its manifest's emits), and -// the forge module consumes it and writes the status (mesh-catalog modules/gitea check.ts). -const CheckStatusEvent = "check.status" - -// CheckStatus is the body of a check.status: the status the checker owes on a head, the owed key and the revision it was -// owed at, by which the forge module drops a stale one delivered after a newer. -type CheckStatus struct { - Owner string `json:"owner"` - Repo string `json:"repo"` - Sha string `json:"sha"` - State string `json:"state"` - Description string `json:"description"` - TargetURL string `json:"target_url,omitempty"` - Key string `json:"key"` - Revision uint64 `json:"revision"` -} - -var wholeSha = regexp.MustCompile(`^([0-9a-f]{40}|[0-9a-f]{64})$`) - -// CheckStatusMessage turns the checker's owed status (mesh-sdk go/delivery CheckerOwedKey, EncodeCheckerOwed) at revision rev -// into the check.status it is emitted as: its id, `check..`, the bus's dedup id too, and its body. A key that is -// not the checker's, a value that is not its status, and a head that is not a whole sha are errors, each saying why. -func CheckStatusMessage(key string, rev uint64, value []byte) (id string, body []byte, err error) { - repository, _, head, err := delivery.ParseCheckerOwedKey(key) - if err != nil { - return "", nil, err - } - o, err := delivery.DecodeCheckerOwed(value) - if err != nil { - return "", nil, fmt.Errorf("the checker's owed status %s: %w", key, err) - } - if o.Act != delivery.OwedStatus || o.Context != delivery.StatusContextCheck { - return "", nil, fmt.Errorf("the checker's owed %s is %s %s, not the status %s", key, o.Act, o.Context, delivery.StatusContextCheck) - } - if !wholeSha.MatchString(head) { - return "", nil, fmt.Errorf("the checker's owed %s names the head %q, not a whole sha", key, head) - } - owner, repo := ownerRepo(repository) - body, err = json.Marshal(CheckStatus{Owner: owner, Repo: repo, Sha: head, State: o.State, Description: o.Text, - TargetURL: o.Target, Key: key, Revision: rev}) - if err != nil { - return "", nil, err - } - return fmt.Sprintf("check.%s.%d", key, rev), body, nil -} - // MergeAct is the name of the act of merging a pull request. const MergeAct = "merge" @@ -128,8 +78,10 @@ func OwedAct(key string, value []byte) (Act, error) { return Act{}, fmt.Errorf("the owed status %s says the context %q, not the %q its key names", key, o.Context, context) } if o.Context == delivery.StatusContextCheck { - return Act{}, fmt.Errorf("the owed status %s is %s, which no tool writes: the bridge emits it as %s (CheckStatusMessage)", - key, delivery.StatusContextCheck, CheckStatusEvent) + // No tool writes mesh/check (R4e finding K1), and the bridge performs no act of it: the forge module writes it from + // the checker's state mesh-delivery-checker.owed, read by a watch (ADR 0521, finding R2-1). + return Act{}, fmt.Errorf("the owed status %s is %s, which the bridge does not perform: the forge module writes it from mesh-delivery-checker.owed", + key, delivery.StatusContextCheck) } a.Call = Call{Tool: commitStatusTool, Args: map[string]any{"owner": owner, "repo": repo, "sha": head, "context": o.Context, "state": o.State, "description": o.Text, "target_url": o.Target}} diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go index 447a679..4883012 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go @@ -58,49 +58,23 @@ func checkerKV(t *testing.T, o delivery.Owed) (string, []byte) { return key, v } -// R4e finding K1 (issue 1301 S4; ADR 0503 §3): no tool writes mesh/check, since a tool is told no caller and node-tools -// invokes every tool. The bridge emits check.status, a subject only its own account may publish, and gitea writes it. +// R4e findings K1 and R2-1 (issue 1301 S4; ADR 0503 §3, ADR 0521): no tool writes mesh/check, since a tool is told no +// caller and node-tools invokes every tool; the forge module writes it from the checker's owed state, and the bridge +// performs no mesh/check act. func TestOwedMeshCheckIsNoToolCall(t *testing.T) { // K1 key, v := checkerKV(t, delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.1"}) a, err := OwedAct(key, v) - if err == nil || !strings.Contains(err.Error(), "check.status") { - t.Fatalf("OwedAct(mesh/check) = %+v, %v; want refused, naming check.status", a, err) + if err == nil || !strings.Contains(err.Error(), "mesh-delivery-checker.owed") { + t.Fatalf("OwedAct(mesh/check) = %+v, %v; want refused, naming where the forge module reads it", a, err) } // Nor through the gate's own owed bucket: the context is refused whichever key carries it. gk, gv := owedValue(t, "novox/mesh-delivery", 12, "status=mesh/check", delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://x.example"}) - if a, err := OwedAct(gk, gv); err == nil || !strings.Contains(err.Error(), "check.status") { + if a, err := OwedAct(gk, gv); err == nil || !strings.Contains(err.Error(), "mesh-delivery-checker.owed") { t.Fatalf("OwedAct(gate's mesh/check) = %+v, %v; want refused", a, err) } } -func TestCheckStatusMessageCarriesTheOwedStatusAndItsRevision(t *testing.T) { // K1 - key, v := checkerKV(t, delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.1"}) - id, body, err := CheckStatusMessage(key, 42, v) - if err != nil { - t.Fatal(err) - } - if CheckStatusEvent != "check.status" { - t.Errorf("event = %q", CheckStatusEvent) - } - if id != "check."+key+".42" { - t.Errorf("id = %q", id) - } - want := `{"owner":"novox","repo":"mesh-delivery","sha":"` + headA + `","state":"success","description":"pass","target_url":"https://view.example/checks/g.1","key":"` + key + `","revision":42}` - if string(body) != want { - t.Errorf("body = %s\nwant %s", body, want) - } - // A key that is not the checker's, and a value that is not its status, are no message. - gk, gv := owedValue(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"}) - if _, _, err := CheckStatusMessage(gk, 1, gv); err == nil { - t.Error("a key of the gate made a check.status") - } - if _, _, err := CheckStatusMessage(key, 1, []byte("{")); err == nil { - t.Error("a value that is no owed status made a check.status") - } -} - func TestOwedViewNoteAndCommentCalls(t *testing.T) { // 1341/AC2 cases := []struct { act string diff --git a/modules/mesh-delivery-forge-bridge/manifest_test.go b/modules/mesh-delivery-forge-bridge/manifest_test.go index 052446d..75e356b 100644 --- a/modules/mesh-delivery-forge-bridge/manifest_test.go +++ b/modules/mesh-delivery-forge-bridge/manifest_test.go @@ -45,7 +45,6 @@ type manifest struct { State []state `json:"state"` Reads []string `json:"reads"` Invokes []string `json:"invokes"` - Emits []string `json:"emits"` OwnSecrets map[string]string `json:"own-secrets"` SecretsOwner string `json:"secrets-owner"` Resources []map[string]any `json:"resources"` @@ -87,16 +86,13 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1 if !reflect.DeepEqual(m.Streams, []stream{{Name: "forge", Events: []string{"forge.**"}}}) { t.Errorf("streams = %+v", m.Streams) } - if want := []string{"gitea.pull.closed", "gitea.pull.merged", "gitea.pull.updated", "mesh-delivery-checker.owed.written"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + // The checker's owed is the forge module's to write (ADR 0521, R4e finding R2-1): the bridge neither hears nor reads it. + if want := []string{"gitea.pull.closed", "gitea.pull.merged", "gitea.pull.updated"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } - if want := []string{"mesh-delivery-checker.owed", "mesh-delivery-gate.merges", "mesh-delivery-gate.owed"}; !reflect.DeepEqual(sorted(m.Reads), want) { + if want := []string{"mesh-delivery-gate.merges", "mesh-delivery-gate.owed"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } - // No tool writes mesh/check (R4e finding K1): the bridge emits check.status, which only its account may publish. - if want := []string{"check.status"}; !reflect.DeepEqual(m.Emits, want) { - t.Errorf("emits = %v, want %v", m.Emits, want) - } if want := []string{"gitea.gitea_add_comment", "gitea.gitea_commit_status", "gitea.gitea_commit_trees", "gitea.gitea_delivery_view", "gitea.gitea_get_file", "gitea.gitea_merge_pull_request", "gitea.gitea_note_append", "gitea.gitea_open_pulls"}; !reflect.DeepEqual(sorted(m.Invokes), want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) @@ -113,8 +109,8 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1 if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}}) { // 1279/AC6, 1341 t.Errorf("state = %+v, want the buckets heads and performed", m.State) } - // No other state, no bus, no seat, no tool (ADR 0437 §4); its one event is check.status, asserted above (R4e finding K1). - for _, k := range []string{"logs", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} { + // No other state, no bus, no seat, no tool, no event (ADR 0437 §4; ADR 0521: no check.status). + for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the bridge has none", k) } diff --git a/modules/mesh-delivery-forge-bridge/module.json b/modules/mesh-delivery-forge-bridge/module.json index c8e5f92..25b68be 100644 --- a/modules/mesh-delivery-forge-bridge/module.json +++ b/modules/mesh-delivery-forge-bridge/module.json @@ -5,8 +5,7 @@ "consumes": [ "gitea.pull.updated", "gitea.pull.merged", - "gitea.pull.closed", - "mesh-delivery-checker.owed.written" + "gitea.pull.closed" ], "streams": [ { "name": "forge", "events": ["forge.**"] } @@ -16,13 +15,9 @@ { "name": "performed", "history": 1 } ], "reads": [ - "mesh-delivery-checker.owed", "mesh-delivery-gate.merges", "mesh-delivery-gate.owed" ], - "emits": [ - "check.status" - ], "invokes": [ "gitea.gitea_merge_pull_request", "gitea.gitea_commit_status",