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 6262776..0a595c7 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 @@ -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 _ 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..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") + } +} 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 b896c7e..68e70ac 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 @@ -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 diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open.go new file mode 100644 index 0000000..0d13d24 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open.go @@ -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) + } + } + } +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open_test.go new file mode 100644 index 0000000..33fb81d --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open_test.go @@ -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") +} diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index a75a991..4b450de 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -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 +} diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go index 981ba50..836ba0b 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go @@ -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) + } + } +}