Say the forge's open pull requests on FORGE at start and at each reconnect (issue 1237)

2026-10-19 14:56:17 +02:00
parent 455b902332
commit 1fdeef91ac
6 changed files with 342 additions and 6 deletions
@@ -9,6 +9,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"

@@ -66,7 +67,7 @@ func startBus(t *testing.T) *server.Server {
 // startBridge runs the program against an embedded bus with the stream EVENTS, the bridge's stream FORGE and its durable
 // consumer <node>_<module> on EVENTS, as the controller makes them (mesh-control internal/broker streams.go:192,
 // nats.go:1387-1396, modstreams.go). It returns a control connection, its JetStream and a context.
-func startBridge(t *testing.T) (*nats.Conn, jetstream.JetStream, context.Context) {
+func startBridge(t *testing.T, openReply string) (*nats.Conn, jetstream.JetStream, context.Context, <-chan error) {
 	srv := startBus(t)
 	ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw"))
 	if err != nil {
@@ -92,6 +93,17 @@ func startBridge(t *testing.T) (*nats.Conn, jetstream.JetStream, context.Context
 		t.Fatal(err)
 	}

+	// A stand-in answers the forge's gitea_open_pulls, which the bridge asks at its start, on its tool subject
+	// (mesh-control internal/broker/nats.go invokedSubjects) in the reply envelope of mesh-sdk go/process/call.go.
+	open, err := ctl.Subscribe("mesh.mod.gitea.tool.gitea_open_pulls", func(m *nats.Msg) { _ = m.Respond([]byte(openReply)) })
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = open.Unsubscribe() })
+	if err := ctl.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
 	broker := filepath.Join(t.TempDir(), "broker")
 	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-forge-bridge", User: "n1.mesh-delivery-forge-bridge", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
@@ -101,13 +113,13 @@ func startBridge(t *testing.T) (*nats.Conn, jetstream.JetStream, context.Context
 	runCtx, stop := context.WithCancel(ctx)
 	done := make(chan error, 1)
 	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
-	t.Cleanup(func() { stop(); <-done })
-	return ctl, js, ctx
+	t.Cleanup(stop)
+	return ctl, js, ctx, done
 }

 // TestBridgeWritesPullMessages starts the program the way the node does (its entry, run, with MESH_BROKER_FILE).
 func TestBridgeWritesPullMessages(t *testing.T) { // 1235/AC2
-	_, js, ctx := startBridge(t)
+	_, js, ctx, _ := startBridge(t, `{"result":{"pulls":[]}}`)

 	// gitea.pull.updated as the forge module publishes it: the body of trunk.ts:233-250 on mesh.mod.gitea.event.pull.updated.
 	ev, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-delivery", "number": 12, "title": "t", "body": "b", "base": "main",
@@ -165,7 +177,7 @@ const merge = "89abcdef0123456789abcdef0123456789abcdef"
 // (mesh-control internal/broker/nats.go invokedSubjects: mesh.mod.<module>.tool.<tool>), in the reply envelope of
 // mesh-sdk go/process/call.go ({result, error, node}).
 func TestBridgeWritesMergedMessage(t *testing.T) { // 1236/AC4
-	ctl, js, ctx := startBridge(t)
+	ctl, js, ctx, _ := startBridge(t, `{"result":{"pulls":[]}}`)
 	var asked chan []byte = make(chan []byte, 4)
 	sub, err := ctl.Subscribe("mesh.mod.gitea.tool.gitea_commit_trees", func(m *nats.Msg) {
 		asked <- m.Data
@@ -238,3 +250,46 @@ func TestBridgeWritesMergedMessage(t *testing.T) { // 1236/AC4
 	ci, _ := cons.Info(ctx)
 	t.Errorf("the event is not acknowledged: %+v", ci)
 }
+
+// TestBridgeSaysTheOpenPullsAtStart: the forge holds two open pull requests when the bridge starts; one message each is on FORGE.
+func TestBridgeSaysTheOpenPullsAtStart(t *testing.T) { // 1237/AC1
+	// The entries are gitea_open_pulls's (mesh-catalog modules/gitea/client.ts:369-381, head the branch).
+	_, js, ctx, _ := startBridge(t, `{"result":{"pulls":[
+	 {"owner":"novox","repo":"mesh-delivery","number":12,"base":"main","head":"feat/x","head_sha":"`+head+`","html_url":"h","required":[],"statuses":[],"checks":{},"labels":[]},
+	 {"owner":"novox","repo":"mesh-catalog","number":300,"base":"main","head":"feat/x","head_sha":"`+strings.Repeat("a", 40)+`","html_url":"h","required":[],"statuses":[],"checks":{},"labels":[]}]},"node":"n0"}`)
+	forge, err := js.Stream(ctx, "FORGE")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var info *jetstream.StreamInfo
+	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 2 {
+			break
+		}
+	}
+	if info.State.Msgs != 2 {
+		t.Fatalf("FORGE holds %d messages within 5 s, want 2", info.State.Msgs)
+	}
+	for _, repo := range []string{"mesh-delivery", "mesh-catalog"} {
+		raw, err := forge.GetLastMsgForSubject(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox."+repo+".pull")
+		if err != nil {
+			t.Fatal(err)
+		}
+		if f, err := delivery.DecodeForge(raw.Data); err != nil || f.Action != "updated" || f.DirsKnown || f.Branch != "feat/x" {
+			t.Errorf("%s: %+v err %v", repo, f, err)
+		}
+	}
+}
+
+// TestBridgeExitsWhenTheOpenPullsCannotBeRead: the process manager starts it again.
+func TestBridgeExitsWhenTheOpenPullsCannotBeRead(t *testing.T) { // 1237/AC3
+	_, _, _, done := startBridge(t, `{"error":"the forge said 502"}`)
+	select {
+	case err := <-done:
+		if err == nil || !strings.Contains(err.Error(), "gitea.gitea_open_pulls") || !strings.Contains(err.Error(), "the forge said 502") {
+			t.Errorf("run returned %v", err)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("the bridge did not exit")
+	}
+}
@@ -157,12 +157,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
+	again := make(chan struct{}, 1)
 	opts := []nats.Option{
 		nats.Name(cred.Node + "." + cred.Module),
 		nats.UserInfo(cred.User, cred.Password),
 		nats.CustomInboxPrefix("_INBOX." + cred.User),
 		nats.MaxReconnects(-1),
 		nats.RetryOnFailedConnect(true),
+		nats.ReconnectHandler(func(*nats.Conn) {
+			select {
+			case again <- struct{}{}:
+			default:
+			}
+		}),
 	}
 	if strings.TrimSpace(cred.Fingerprint) != "" {
 		opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
@@ -176,11 +183,18 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
+	// What the forge holds open, said before any event is heard, and again at each reconnect (ADR 0429 §13).
+	call := callerOn(conn, cred.User)
+	sync := func(ctx context.Context) error { return syncOpen(ctx, call, js, logger) }
+	if err := sync(ctx); err != nil {
+		return err
+	}
+	go onReconnect(ctx, again, sync, logger)
 	cons, err := js.Consumer(ctx, eventsStream, cred.Node+"_"+cred.Module)
 	if err != nil {
 		return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, eventsStream, err)
 	}
-	cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, js, treesVia(ctx, callerOn(conn, cred.User)), logger) },
+	cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, js, treesVia(ctx, call), logger) },
 		jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
 	if err != nil {
 		return err
@@ -0,0 +1,52 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"mesh-delivery-forge-bridge/internal/bridge"
+)
+
+// openPullsTool is the forge's tool for every open pull request (mesh-catalog modules/gitea/tools/index.ts
+// gitea_open_pulls: no arguments, answers {pulls: [...]}).
+const openPullsTool = "gitea.gitea_open_pulls"
+
+// syncOpen reads the open pull requests from the forge and writes one pull message for each: what makes the indexer's
+// candidates match the forge after a restart or a gap in the forge's events (ADR 0429 §13). An error names the call and
+// the forge's words.
+func syncOpen(ctx context.Context, call caller, pub publisher, logger *log.Logger) error {
+	raw, err := call(ctx, openPullsTool, struct{}{})
+	if err != nil {
+		return fmt.Errorf("%s: %w", openPullsTool, err)
+	}
+	msgs, skipped, err := bridge.OpenPulls(raw)
+	if err != nil {
+		return fmt.Errorf("%s: %w", openPullsTool, err)
+	}
+	for _, s := range skipped {
+		logger.Printf("an open pull request is left out: %s", s)
+	}
+	for _, m := range msgs {
+		if _, err := pub.Publish(ctx, forgeSubject(m.Event), m.Data); err != nil {
+			return fmt.Errorf("the open pull request message %s could not be written on FORGE: %w", m.Event, err)
+		}
+	}
+	logger.Printf("%d open pull requests said on FORGE", len(msgs))
+	return nil
+}
+
+// onReconnect reads the open pull requests again each time the bus connection comes back, until ctx ends; a failure is
+// logged and the bridge goes on (the next reconnect or event tries again).
+func onReconnect(ctx context.Context, again <-chan struct{}, sync func(context.Context) error, logger *log.Logger) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-again:
+			if err := sync(ctx); err != nil {
+				logger.Printf("after a reconnect: %v", err)
+			}
+		}
+	}
+}
@@ -0,0 +1,114 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"log"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// openAnswer is gitea_open_pulls's answer as mesh-catalog modules/gitea/client.ts:369-381 builds its entries inside {pulls}.
+func openAnswer() json.RawMessage {
+	return json.RawMessage(`{"pulls":[
+	 {"owner":"novox","repo":"mesh-delivery","number":12,"base":"main","head":"feat/x","head_sha":"` + head + `","html_url":"h","required":[],"statuses":[],"checks":{},"labels":[]},
+	 {"owner":"novox","repo":"mesh-catalog","number":300,"base":"main","head":"feat/x","head_sha":"` + strings.Repeat("a", 40) + `","html_url":"h","required":[],"statuses":[],"checks":{},"labels":[]},
+	 {"owner":"novox","repo":"mesh-catalog","number":301,"base":"feat/y","head":"feat/z","head_sha":"` + strings.Repeat("b", 40) + `","html_url":"h","required":[],"statuses":[],"checks":{},"labels":[]}]}`)
+}
+
+func TestSyncOpenWritesOneUpdatedMessagePerOpenPull(t *testing.T) { // 1237/AC1, AC4
+	var logs bytes.Buffer
+	pub := &fakePub{}
+	var keys []string
+	call := func(_ context.Context, key string, _ any) (json.RawMessage, error) {
+		keys = append(keys, key)
+		return openAnswer(), nil
+	}
+	if err := syncOpen(context.Background(), call, pub, log.New(&logs, "", 0)); err != nil {
+		t.Fatal(err)
+	}
+	if len(keys) != 1 || keys[0] != "gitea.gitea_open_pulls" {
+		t.Errorf("calls %v", keys)
+	}
+	if len(pub.subjects) != 2 {
+		t.Fatalf("%d messages written (%v), want two: the pull request into feat/y gets none", len(pub.subjects), pub.subjects)
+	}
+	want := []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull", "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-catalog.pull"}
+	for i, w := range want {
+		f, err := delivery.DecodeForge(pub.data[i])
+		if pub.subjects[i] != w || err != nil || f.Action != "updated" || f.DirsKnown || f.Branch != "feat/x" {
+			t.Errorf("message %d on %s = %+v err %v", i, pub.subjects[i], f, err)
+		}
+	}
+}
+
+func TestSyncOpenFailureNamesTheCallAndTheForgesWords(t *testing.T) { // 1237/AC3
+	pub := &fakePub{}
+	call := func(context.Context, string, any) (json.RawMessage, error) {
+		return nil, errors.New("the forge said 502")
+	}
+	err := syncOpen(context.Background(), call, pub, log.New(&bytes.Buffer{}, "", 0))
+	if err == nil || !strings.Contains(err.Error(), "gitea.gitea_open_pulls") || !strings.Contains(err.Error(), "the forge said 502") {
+		t.Errorf("err = %v", err)
+	}
+	if len(pub.subjects) != 0 {
+		t.Errorf("messages written: %v", pub.subjects)
+	}
+}
+
+func TestReconnectReadsTheOpenPullsAgainAndFailureGoesOn(t *testing.T) { // 1237/AC2, AC3
+	var logs lockedBuffer
+	ctx, cancel := context.WithCancel(context.Background())
+	again := make(chan struct{}, 1)
+	done := make(chan struct{})
+	var mu sync.Mutex
+	n := 0
+	count := func() int { mu.Lock(); defer mu.Unlock(); return n }
+	read := func(context.Context) error {
+		mu.Lock()
+		defer mu.Unlock()
+		n++
+		if n == 1 {
+			return errors.New("the forge said 502")
+		}
+		return nil
+	}
+	go func() { onReconnect(ctx, again, read, log.New(&logs, "", 0)); close(done) }()
+	again <- struct{}{}
+	waitFor(t, func() bool { return count() >= 1 })
+	again <- struct{}{} // the first read failed; the bridge goes on and reads again at the next reconnect
+	waitFor(t, func() bool { return count() >= 2 })
+	cancel()
+	<-done
+	if !strings.Contains(logs.String(), "the forge said 502") {
+		t.Errorf("log %q does not carry the forge's words", logs.String())
+	}
+}
+
+type lockedBuffer struct {
+	mu sync.Mutex
+	b  bytes.Buffer
+}
+
+func (l *lockedBuffer) Write(p []byte) (int, error) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.b.Write(p)
+}
+func (l *lockedBuffer) String() string { l.mu.Lock(); defer l.mu.Unlock(); return l.b.String() }
+
+func waitFor(t *testing.T, ok func() bool) {
+	t.Helper()
+	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(5 * time.Millisecond) {
+		if ok() {
+			return
+		}
+	}
+	t.Fatal("it did not happen within 5 s")
+}
@@ -139,3 +139,50 @@ func merged(subject string, p pull, trees Trees) (Message, error) {
 	}
 	return Message{Event: event, Data: body}, nil
 }
+
+// openPull is an entry of gitea_open_pulls's answer (mesh-catalog modules/gitea/client.ts:369-381 openPulls, with the
+// branch as `head`, issue 1234/AC5), the part the bridge reads.
+type openPull struct {
+	Owner   string `json:"owner"`
+	Repo    string `json:"repo"`
+	Number  int    `json:"number"`
+	Base    string `json:"base"`
+	Head    string `json:"head"`
+	HeadSHA string `json:"head_sha"`
+}
+
+// OpenPulls turns the answer of gitea_open_pulls, {pulls: [...]}, into one pull message per open pull request into
+// the trunk, action updated and its directories not known (the forge's answer does not say them). A pull request into
+// another branch gets none. An entry that cannot be read is named in skipped and gets none; an answer that is not
+// {pulls} is an error.
+func OpenPulls(answer []byte) (msgs []Message, skipped []string, err error) {
+	var a struct {
+		Pulls *[]openPull `json:"pulls"`
+	}
+	if err := json.Unmarshal(answer, &a); err != nil || a.Pulls == nil {
+		return nil, nil, fmt.Errorf("the answer of gitea_open_pulls is not {pulls: [...]}: %v", err)
+	}
+	for _, p := range *a.Pulls {
+		if p.Base != Trunk {
+			continue
+		}
+		name := fmt.Sprintf("%s/%s#%d", p.Owner, p.Repo, p.Number)
+		if p.Owner == "" || p.Repo == "" || p.Number <= 0 || p.Head == "" || !sha40.MatchString(p.HeadSHA) {
+			skipped = append(skipped, name+": it names no owner, repo, number, branch or 40-hex head")
+			continue
+		}
+		f := delivery.Forge{Kind: delivery.ForgePull, Repository: p.Owner + "/" + p.Repo, Branch: p.Head, Number: p.Number,
+			Head: p.HeadSHA, Action: delivery.PullUpdated}
+		event, err := delivery.ForgeEvent(f.Repository, delivery.ForgePull)
+		var body []byte
+		if err == nil {
+			body, err = delivery.EncodeForge(f)
+		}
+		if err != nil {
+			skipped = append(skipped, name+": "+err.Error())
+			continue
+		}
+		msgs = append(msgs, Message{Event: event, Data: body})
+	}
+	return msgs, skipped, nil
+}
@@ -269,3 +269,57 @@ func TestMergedWithoutAMergeCommitIsUnreadable(t *testing.T) { // 1236
 		t.Errorf("err = %v calls = %v", err, calls)
 	}
 }
+
+// openAnswer is gitea_open_pulls's answer as mesh-catalog modules/gitea/client.ts:369-381 builds each entry
+// (owner, repo, number, base, head, head_sha, html_url, required, statuses, checks, labels) inside {pulls}.
+func openAnswer(pulls ...map[string]any) []byte {
+	b, _ := json.Marshal(map[string]any{"pulls": pulls})
+	return b
+}
+
+func openEntry(repo string, number int, base, branch, sha string) map[string]any {
+	return map[string]any{"owner": "novox", "repo": repo, "number": number, "base": base, "head": branch, "head_sha": sha,
+		"html_url": "https://forge.example/x", "required": []string{"mesh/merge-gate"}, "statuses": []string{}, "checks": map[string]string{}, "labels": []string{}}
+}
+
+func TestOpenPullsAreUpdatedPullMessagesWithUnknownDirs(t *testing.T) { // 1237/AC1
+	h2 := strings.Repeat("a", 40)
+	msgs, skipped, err := OpenPulls(openAnswer(openEntry("mesh-delivery", 12, "main", "feat/x", head), openEntry("mesh-catalog", 300, "main", "feat/x", h2)))
+	if err != nil || len(skipped) != 0 || len(msgs) != 2 {
+		t.Fatalf("msgs %d skipped %v err %v", len(msgs), skipped, err)
+	}
+	for i, want := range []delivery.Forge{
+		{V: delivery.Version, Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullUpdated},
+		{V: delivery.Version, Kind: delivery.ForgePull, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 300, Head: h2, Action: delivery.PullUpdated},
+	} {
+		f, err := delivery.DecodeForge(msgs[i].Data)
+		if err != nil || !delivery.SameForge(f, want) {
+			t.Errorf("message %d = %+v err %v, want %+v", i, f, err, want)
+		}
+		if f.DirsKnown {
+			t.Errorf("message %d says its directories are known", i)
+		}
+	}
+	if msgs[0].Event != "forge.novox.mesh-delivery.pull" || msgs[1].Event != "forge.novox.mesh-catalog.pull" {
+		t.Errorf("events %q %q", msgs[0].Event, msgs[1].Event)
+	}
+}
+
+func TestOpenPullIntoAnotherBranchGetsNoMessage(t *testing.T) { // 1237/AC4
+	msgs, skipped, err := OpenPulls(openAnswer(openEntry("mesh-delivery", 12, "feat/y", "feat/x", head)))
+	if err != nil || len(msgs) != 0 || len(skipped) != 0 {
+		t.Errorf("msgs %d skipped %v err %v, want none", len(msgs), skipped, err)
+	}
+}
+
+func TestOpenPullsUnreadableEntryIsNamedAndTheRestGo(t *testing.T) { // 1237
+	msgs, skipped, err := OpenPulls(openAnswer(openEntry("a", 1, "main", "b", "short"), openEntry("mesh-delivery", 12, "main", "feat/x", head)))
+	if err != nil || len(msgs) != 1 || len(skipped) != 1 || !strings.Contains(skipped[0], "novox/a#1") {
+		t.Errorf("msgs %d skipped %v err %v", len(msgs), skipped, err)
+	}
+	for _, bad := range []string{`{"pulls":null}`, `{}`, `[]`, `x`} {
+		if _, _, err := OpenPulls([]byte(bad)); err == nil {
+			t.Errorf("%s was taken as an answer", bad)
+		}
+	}
+}