diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go new file mode 100644 index 00000000..f3638d1e --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go @@ -0,0 +1,174 @@ +// Command mesh-delivery-forge-bridge-tools serves the forge bridge's tools (novox/hq ADR 0633 §4, issue 1585): the runtime +// launches it for a call. bridge_groups reads the delivery groups the gate holds and what the bridge kept of their ends; +// bridge_group_end_again writes an ask into the bridge's state end-asks and waits for the bridge's answer there, since the +// bridge is the other program and serves no request subject (ADR 0437 §4). +package main + +import ( + "context" + "encoding/json" + "fmt" + "os" + "strconv" + "strings" + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-forge-bridge/internal/groups" +) + +// Tool names. +const ( + GroupsTool = "bridge_groups" + EndAgainTool = "bridge_group_end_again" +) + +// asker writes an end-asks value and waits for the bridge's answer to it. +type asker func(key string, value []byte) (groups.EndAsk, error) + +// Tools are the bridge's tools: groupsRead reads the groups the gate holds with what the bridge kept of them. +func Tools(ask asker, now func() time.Time, groupsRead func() ([]groups.Row, []string, error)) []stdio.Tool { + str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } + return []stdio.Tool{ + {Name: GroupsTool, + Description: "The delivery groups the gate holds (more than one member), each with its members' merge state and whether each " + + "merged message is written on the forge stream, and whether the group's end (forge.group.merged) is written: ended, " + + "or open with why (gathered, or merged and its end not written yet). An open group with every member merged is " + + "what bridge_group_end_again says the end of. A group's name narrows it to that group; open lists only open ones.", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "group": map[string]any{"type": "string", "description": "only this group, by the name the gate keys it"}, + "open": map[string]any{"type": "boolean", "description": "only the groups whose end is not written"}}}, + Run: func(args map[string]any) (any, error) { + rows, unread, err := groupsRead() + if err != nil { + return nil, err + } + only, _ := args["open"].(bool) + group := str(args, "group") + kept := rows[:0] + for _, r := range rows { + if (group != "" && r.Group != group) || (only && !r.Open) { + continue + } + kept = append(kept, r) + } + return map[string]any{"groups": kept, "unread": unread}, nil + }}, + {Name: EndAgainTool, + Description: "Say a delivery group's end (forge.group.merged) again, by hand, for a group whose end the indexer did not take " + + "in (a group bridge_groups names as merged with its end not written, or one the indexer waits for). The bridge " + + "writes it again from the gate's merges, with the time appended to the message id so the stream does not drop it as " + + "a duplicate; the indexer takes an end for a group it already took in without changing anything. Refused while " + + "a member is not merged or its merged message is not written. Answers whether it was written.", + Input: map[string]any{"type": "object", "required": []string{"group"}, "properties": map[string]any{ + "group": map[string]any{"type": "string", "description": "the group, by the name the gate keys it"}, + "why": map[string]any{"type": "string", "description": "why its end is said again, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + key, value, err := groups.NewEndAsk(str(args, "group"), str(args, "why"), stdio.CallerOf(args), now()) + if err != nil { + return nil, err + } + a, err := ask(key, value) + if err != nil { + return nil, err + } + return map[string]any{"group": a.Group, "written": a.Written, "answer": a.Answer, "by": a.By, "why": a.Why}, nil + }}, + } +} + +// read is the groups the gate holds with what the bridge kept of them. +func read(merges, performed stdio.KeptState) ([]groups.Row, []string, error) { + keys, err := merges.Keys() + if err != nil { + return nil, nil, fmt.Errorf("the gate's merges cannot be listed: %w", err) + } + raws := map[string][]byte{} + for _, k := range keys { + e, err := merges.Get(k) + if err != nil || e == nil { + continue + } + raws[k] = e.Value + } + pkeys, err := performed.Keys() + if err != nil { + return nil, nil, fmt.Errorf("the bridge's performed state cannot be listed: %w", err) + } + published, ended := map[string]bool{}, map[string]uint64{} + for _, k := range pkeys { + switch { + case strings.HasPrefix(k, "published."): + published[k] = true + case strings.HasPrefix(k, "groups."): + if e, err := performed.Get(k); err == nil && e != nil { + rev, _ := strconv.ParseUint(strings.TrimSpace(string(e.Value)), 10, 64) + ended[strings.TrimPrefix(k, "groups.")] = rev + } + } + } + rows, unread := groups.Rows(raws, published, ended) + return rows, unread, nil +} + +// askEnd writes the end-asks value at key and waits, on a watch of the key, at most bound for the bridge's answer. +func askEnd(s stdio.KeptState, bound time.Duration) asker { + return func(key string, value []byte) (groups.EndAsk, error) { + answered := make(chan groups.EndAsk, 1) + var rev uint64 + var mu sync.Mutex + take := func(revision uint64, v []byte) { + mu.Lock() + mine := rev + mu.Unlock() + if mine == 0 || revision <= mine { + return + } + if a, err := groups.DecodeEndAsk(v); err == nil && a.Done() { + select { + case answered <- a: + default: + } + } + } + if err := s.Watch(key, func(c stdio.StateChange) error { + if c.Op == "put" { + take(c.Revision, c.Value) + } + return nil + }); err != nil { + return groups.EndAsk{}, fmt.Errorf("watching %s.%s: %w", groups.EndAsksState, key, err) + } + r, err := s.Put(key, json.RawMessage(value)) + if err != nil { + return groups.EndAsk{}, fmt.Errorf("the ask could not be written to %s: %w", groups.EndAsksBucket, err) + } + mu.Lock() + rev = r + mu.Unlock() + if e, err := s.Get(key); err == nil && e != nil { + take(e.Revision, e.Value) + } + wait, cancel := context.WithTimeout(context.Background(), bound) // a context's bound, no timer (ADR 0437 §8) + defer cancel() + select { + case a := <-answered: + return a, nil + case <-wait.Done(): + return groups.EndAsk{}, fmt.Errorf("the bridge did not answer within %s: its answer will be in %s.%s", bound, groups.EndAsksBucket, key) + } + } +} + +func main() { + merges, performed := stdio.State(delivery.GateModule+".merges"), stdio.State(groups.PerformedState) + tools := Tools(askEnd(stdio.State(groups.EndAsksState), 25*time.Second), time.Now, + func() ([]groups.Row, []string, error) { return read(merges, performed) }) + if err := stdio.Serve("", tools); err != nil { + fmt.Fprintf(os.Stderr, "[mesh-delivery-forge-bridge-tools] %v\n", err) + os.Exit(1) + } +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go new file mode 100644 index 00000000..cdfe4250 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go @@ -0,0 +1,109 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-forge-bridge/internal/groups" +) + +// 1585/T18, T19 (ADR 0633 §4): bridge_groups names each group the gate holds with its members' state and whether its end is +// written (open or ended, with why), narrowed by group and by open; bridge_group_end_again writes an ask naming the group, +// who asked and why, and answers what the bridge answered. +func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) { + h1, h2 := strings.Repeat("1", 40), strings.Repeat("2", 40) + c1, c2 := strings.Repeat("c", 40), strings.Repeat("d", 40) + enc := func(group string, state2 delivery.MergeState, c string) []byte { + v, err := delivery.EncodeMerge(delivery.Merge{Group: group, Ask: "ask", Approved: map[string]string{"novox/a@1": h1, "novox/b@2": h2}, + Pulls: []delivery.MergePull{{Head: delivery.Head{Repository: "novox/a", Number: 1, Head: h1}, State: delivery.MergeMerged, Commit: c1}, + {Head: delivery.Head{Repository: "novox/b", Number: 2, Head: h2}, State: state2, Commit: c}}}) + if err != nil { + t.Fatal(err) + } + return v + } + lone, _ := delivery.EncodeMerge(delivery.Merge{Group: "solo", Ask: "ask", Approved: map[string]string{"novox/a@5": h1}, + Pulls: []delivery.MergePull{{Head: delivery.Head{Repository: "novox/a", Number: 5, Head: h1}, State: delivery.MergeMerged, Commit: c1}}}) + merges := map[string][]byte{"ended": enc("ended", delivery.MergeMerged, c2), "ready": enc("ready", delivery.MergeMerged, c2), + "gathered": enc("gathered", delivery.MergeMerging, ""), "solo": lone, "bad": []byte("{")} + performed := map[string]bool{} + for _, c := range []string{c1, c2} { + performed[groups.PublishedKey("novox/a", 1, c)], performed[groups.PublishedKey("novox/b", 2, c)] = true, true + } + var asked groups.EndAsk + var askedKey string + ask := func(key string, value []byte) (groups.EndAsk, error) { + askedKey = key + a, err := groups.DecodeEndAsk(value) + a.Written, a.Answer, a.Answered = true, "the end of the group is written again", time.Now() + asked = a + return a, err + } + tools := Tools(ask, func() time.Time { return time.Date(2026, 11, 4, 23, 0, 0, 0, time.UTC) }, + func() ([]groups.Row, []string, error) { + rows, unread := groups.Rows(merges, performed, map[string]uint64{"ended": 7}) + return rows, unread, nil + }) + call := func(name string, args map[string]any) any { + for _, tl := range tools { + if tl.Name == name { + out, err := tl.Run(args) + if err != nil { + t.Fatalf("%s: %v", name, err) + } + return out + } + } + t.Fatalf("no tool %s", name) + return nil + } + rowsOf := func(args map[string]any) map[string]groups.Row { + raw, _ := json.Marshal(call(GroupsTool, args)) + var out struct { + Groups []groups.Row `json:"groups"` + Unread []string `json:"unread"` + } + if err := json.Unmarshal(raw, &out); err != nil { + t.Fatal(err) + } + m := map[string]groups.Row{} + for _, r := range out.Groups { + m[r.Group] = r + } + if len(args) == 0 && (len(out.Unread) != 1 || out.Unread[0] != "bad") { + t.Errorf("unread = %v, want bad", out.Unread) + } + return m + } + all := rowsOf(map[string]any{}) + if len(all) != 3 || !all["ended"].Ended || all["ended"].EndedAt != 7 { + t.Fatalf("groups = %+v", all) + } + if r := all["ready"]; !r.Open || !strings.Contains(r.Words, "bridge_group_end_again") || !r.Members[1].Published { + t.Errorf("ready = %+v, want open, merged and published, its end the thing to say", r) + } + if r := all["gathered"]; !r.Open || !strings.Contains(r.Words, "gathered") || r.Members[1].State != string(delivery.MergeMerging) { + t.Errorf("gathered = %+v", r) + } + if open := rowsOf(map[string]any{"open": true}); len(open) != 2 || open["ended"].Group != "" { + t.Errorf("open = %+v, want ready and gathered", open) + } + if one := rowsOf(map[string]any{"group": "ready"}); len(one) != 1 { + t.Errorf("group ready = %+v", one) + } + out := call(EndAgainTool, map[string]any{"group": "ready", "why": "the indexer never took it in"}).(map[string]any) + if askedKey != "ready" || asked.Group != "ready" || asked.Why != "the indexer never took it in" || out["written"] != true { + t.Errorf("the ask %q %+v, answered %v", askedKey, asked, out) + } + for _, tl := range tools { + if tl.Name == EndAgainTool { + if _, err := tl.Run(map[string]any{}); err == nil { + t.Error("an ask naming no group was taken") + } + } + } +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask.go new file mode 100644 index 00000000..3c0365b8 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask.go @@ -0,0 +1,61 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "mesh-delivery-forge-bridge/internal/groups" +) + +// endAsked answers one value of the state end-asks (the tool bridge_group_end_again, ADR 0633 §4): a value not yet answered +// says the group's end again from the gate's merges, with the ask's time appended to the message id so the stream's duplicate +// window does not drop it, and is written back by compare-and-set at its revision with whether it was written. A value +// answered already (this bridge's own write) is left. +func (g *grouper) endAsked(ctx context.Context, kv jetstream.KeyValue, key string, rev uint64, value []byte) { + a, err := groups.DecodeEndAsk(value) + if err != nil { + g.logger.Printf("%s.%s holds what is not an end ask, left: %v", groups.EndAsksState, key, err) + return + } + if a.Done() { + return + } + a = g.answerEnd(ctx, key, a, time.Now()) + g.logger.Printf("the end of the group %q said again by %s (%s): %s", a.Group, a.By, a.Why, a.Answer) + data, err := json.Marshal(a) + if err != nil { + return + } + if _, err := kv.Update(ctx, key, data, rev); err != nil && !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + g.logger.Printf("%s.%s: the answer could not be written: %v", groups.EndAsksState, key, err) + } +} + +// readMerges reads one key of the gate's merges state. +func (g *grouper) readMerges(ctx context.Context, key string) ([]byte, uint64, bool, error) { + if g.merges == nil { + return nil, 0, false, errors.New("the bridge does not read the gate's merges (its bucket was not there at the start)") + } + value, rev, found, err := g.merges(ctx, key) + return value, rev, found, err +} + +// answerEnd says the group's end again for an ask and returns the ask with its answer. +func (g *grouper) answerEnd(ctx context.Context, key string, a groups.EndAsk, at time.Time) groups.EndAsk { + a.Answered = at.UTC() + switch merges, mrev, found, err := g.readMerges(ctx, key); { + case err != nil: + a.Answer = "the gate's merges " + key + " cannot be read: " + err.Error() + case !found: + a.Answer = "the gate holds no merge of the group " + a.Group + case g.end(ctx, key, mrev, merges, groups.AgainStamp(a.At)): + a.Written, a.Answer = true, "the end of the group is written again" + default: + a.Answer = "the end is not written: a member is not merged or its merged message is not written, or FORGE refused it (the bridge's journal says which)" + } + return a +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask_test.go new file mode 100644 index 00000000..3d0bb3af --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask_test.go @@ -0,0 +1,68 @@ +package main + +import ( + "bytes" + "context" + "log" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-forge-bridge/internal/groups" +) + +// 1585/T19 (ADR 0633 §4): an ask to say a group's end again is answered by the bridge from the gate's merges, with the ask's +// time on the message id; not written while a member is not merged, with the reason; not for a group the gate does not hold; +// and the keys of performed the tools read are the ones the process writes. +func TestAnEndAskSaysTheGroupsEndAgainOrWhyNot(t *testing.T) { + ctx := context.Background() + h1, h2 := strings.Repeat("1", 40), strings.Repeat("2", 40) + c1, c2 := strings.Repeat("c", 40), strings.Repeat("d", 40) + pull := func(repo string, n int, head string, state delivery.MergeState, commit string) delivery.MergePull { + return delivery.MergePull{Head: delivery.Head{Repository: repo, Number: n, Head: head}, State: state, Commit: commit} + } + merge := func(second delivery.MergeState, commit2 string) []byte { + v, err := delivery.EncodeMerge(delivery.Merge{Group: "feat/take-in", Ask: "ask", Approved: map[string]string{"novox/mesh-sdk@37": h1, "novox/mesh-delivery@9": h2}, + Pulls: []delivery.MergePull{pull("novox/mesh-sdk", 37, h1, delivery.MergeMerged, c1), pull("novox/mesh-delivery", 9, h2, second, commit2)}}) + if err != nil { + t.Fatal(err) + } + return v + } + ask := groups.EndAsk{V: 1, Group: "feat/take-in", By: "operator@shanks", Why: "the indexer never took it in", At: time.Date(2026, 11, 4, 23, 0, 0, 123, time.UTC)} + for _, tc := range []struct { + name string + gate map[string][]byte + published bool + written bool + answer string + }{ + {"merged and published: written again", map[string][]byte{"feat/take-in": merge(delivery.MergeMerged, c2)}, true, true, "written again"}, + {"a member's message not written", map[string][]byte{"feat/take-in": merge(delivery.MergeMerged, c2)}, false, false, "not written"}, + {"a member not merged", map[string][]byte{"feat/take-in": merge(delivery.MergeMerging, "")}, true, false, "not written"}, + {"no merge at the gate", map[string][]byte{}, true, false, "holds no merge"}, + } { + pub, done := &idPub{}, memRevisions{} + if tc.published { + done[publishedKey("novox/mesh-sdk", 37, c1)], done[publishedKey("novox/mesh-delivery", 9, c2)] = 1, 1 + } + g := &grouper{send: pub.send, done: done, logger: log.New(&bytes.Buffer{}, "", 0), + merges: func(_ context.Context, key string) ([]byte, uint64, bool, error) { + v, ok := tc.gate[key] + return v, 4, ok, nil + }} + got := g.answerEnd(ctx, "feat/take-in", ask, ask.At.Add(time.Second)) + if got.Written != tc.written || !strings.Contains(got.Answer, tc.answer) || !got.Done() || len(pub.ids) != map[bool]int{true: 1}[tc.written] { + t.Errorf("%s: %+v, ids %v", tc.name, got, pub.ids) + } + if tc.written && pub.ids[0] != groupEndID("feat/take-in", c2)+"."+groups.AgainStamp(ask.At) { + t.Errorf("%s: the id %s does not carry the ask's time", tc.name, pub.ids[0]) + } + } + if groupsKey("feat/take-in") != groups.Key("feat/take-in") || publishedKey("novox/mesh-sdk", 37, c1) != groups.PublishedKey("novox/mesh-sdk", 37, c1) || + groupEndID("g", c1) != groups.EndID("g", c1, "") { + t.Error("the tools bundle's keys of performed are not the process's") + } +} 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 c630e424..26331490 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 @@ -24,6 +24,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "mesh-delivery-forge-bridge/internal/bridge" + groupkeys "mesh-delivery-forge-bridge/internal/groups" ) // eventsStream is the shared stream every module's events are on; the module's consumer on it is the durable @@ -177,6 +178,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro }) }() } + // A person says a group's end again with bridge_group_end_again, which writes the state end-asks (ADR 0633 §4): watched, and + // answered there; a state not there at this start is answered from the next start. + if kv := process.OptionalBucket(ctx, js, groupkeys.EndAsksBucket, "bridge_group_end_again is not answered until the bridge starts again", logger.Printf); kv != nil { + go func() { + err := watch(ctx, kv, func(key string, rev uint64, value []byte) { groups.endAsked(ctx, kv, key, rev, value) }) + if err != nil && ctx.Err() == nil { + logger.Printf("the end asks are no longer watched, so bridge_group_end_again is not answered until the bridge starts again: %v", err) + } + }() + } if err := said(ctx); err != nil { return err } diff --git a/modules/mesh-delivery-forge-bridge/internal/groups/groups.go b/modules/mesh-delivery-forge-bridge/internal/groups/groups.go new file mode 100644 index 00000000..006e1703 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/internal/groups/groups.go @@ -0,0 +1,158 @@ +// Package groups is what the forge bridge's process and its tools bundle share of a delivery group's end (novox/hq ADR +// 0633 §4, issue 1585): the keys of the bridge's state performed, and the asks of the state end-asks, through which the +// tool bridge_group_end_again asks the process to say a group's end again (the bridge serves no request subject, ADR 0437 +// §4, so a tool asks in a state the process watches, as indexer_take_in asks the indexer in takeins). +package groups + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The bridge's states the tools read and write. +const ( + // PerformedState is the bridge's state of what it performed and published. + PerformedState = "performed" + // EndAsksState is the state of asks to say a group's end again, keyed by the gate's merges key of the group. + EndAsksState = "end-asks" + // EndAsksBucket is its bucket on the bus. + EndAsksBucket = delivery.ForgeBridgeModule + "_" + EndAsksState +) + +// Key is the key of performed that says a group's end is written: `groups.`, its value the revision of the +// gate's merges key it was worked out from. +func Key(mergesKey string) string { return "groups." + mergesKey } + +// PublishedKey is the key of performed that says a member's merged message is written: `published.<16 hex digits of the +// sha256 of #@>`. +func PublishedKey(repository string, number int, merge string) string { + sum := sha256.Sum256([]byte(fmt.Sprintf("%s#%d@%s", strings.ToLower(repository), number, merge))) + return "published." + hex.EncodeToString(sum[:8]) +} + +// EndID is the message id of a group's end, and of one said again by hand at a time: the stream's duplicate window drops a +// second publish of the same id. +func EndID(group, lastMerge, again string) string { + id := "group-end." + group + "." + lastMerge + if again != "" { + id += "." + again + } + return id +} + +// AgainStamp is the time a hand ask appends to a group end's message id. +func AgainStamp(at time.Time) string { return at.UTC().Format("20060102T150405.000000000Z") } + +// EndAsk is a value of the end-asks state: which group's end a person asked to be said again, why, who and when, and, once +// the bridge read it, its answer. +type EndAsk struct { + V int `json:"v"` + Group string `json:"group"` + Why string `json:"why,omitempty"` + By string `json:"by,omitempty"` + At time.Time `json:"at"` + Written bool `json:"written,omitempty"` + Answer string `json:"answer,omitempty"` + Answered time.Time `json:"answered,omitzero"` +} + +// Done says the ask is answered. +func (a EndAsk) Done() bool { return !a.Answered.IsZero() } + +// NewEndAsk is the value bridge_group_end_again writes for a group: its key (the gate's merges key) and the value. +func NewEndAsk(group, why, by string, at time.Time) (string, []byte, error) { + group = strings.TrimSpace(group) + if group == "" { + return "", nil, errors.New("name the group whose end is said again (group)") + } + key, err := delivery.MergesKey(group) + if err != nil { + return "", nil, err + } + data, err := json.Marshal(EndAsk{V: 1, Group: group, Why: strings.TrimSpace(why), By: by, At: at.UTC()}) + return key, data, err +} + +// DecodeEndAsk reads an end-asks value. +func DecodeEndAsk(data []byte) (EndAsk, error) { + var a EndAsk + if err := json.Unmarshal(data, &a); err != nil { + return a, err + } + if a.V != 1 || a.Group == "" { + return a, errors.New("an end-asks value of no version 1 naming no group") + } + return a, nil +} + +// Row is one group as bridge_groups answers it: the gate's merge of it and what the bridge kept of its end. +type Row struct { + Key string `json:"key"` + Group string `json:"group"` + Members []RowMember `json:"members"` + Ended bool `json:"ended"` + EndedAt uint64 `json:"ended-at-revision,omitempty"` + Open bool `json:"open"` + Words string `json:"words"` +} + +// RowMember is one approved member of a group: its merge's state at the gate and whether its merged message is written. +type RowMember struct { + Repository string `json:"repository"` + Number int `json:"number"` + State string `json:"state"` + Commit string `json:"commit,omitempty"` + Published bool `json:"published"` +} + +// Rows are the groups the gate's merges hold with more than one member, each with what the bridge kept: ended (its end +// written, the revision it was worked out from) or open (some member is not merged or its merged message not written, or its +// end not written). merges is the gate's merges by key, performed the bridge's performed keys by key; unread are the gate's +// keys that do not read. +func Rows(merges map[string][]byte, performed map[string]bool, endedAt map[string]uint64) (rows []Row, unread []string) { + for key, raw := range merges { + m, err := delivery.DecodeMerge(raw) + if err != nil { + unread = append(unread, key) + continue + } + if len(m.Approved) < 2 { + continue + } + r := Row{Key: key, Group: m.Group} + allMerged := true + for _, p := range m.Pulls { + pub := p.Commit != "" && performed[PublishedKey(p.Repository, p.Number, p.Commit)] + r.Members = append(r.Members, RowMember{Repository: p.Repository, Number: p.Number, State: string(p.State), Commit: p.Commit, Published: pub}) + if p.State != delivery.MergeMerged || !pub { + allMerged = false + } + } + if len(m.Pulls) < len(m.Approved) { + allMerged = false + } + r.EndedAt, r.Ended = endedAt[key] + r.Open = !r.Ended + switch { + case r.Ended: + r.Words = "its end is written" + case allMerged: + r.Words = "every member is merged and its message written, and its end is not written: bridge_group_end_again says it" + default: + r.Words = "gathered: not every member is merged and its merged message written" + } + rows = append(rows, r) + } + sortRows(rows) + return rows, unread +} + +func sortRows(rows []Row) { sort.Slice(rows, func(i, j int) bool { return rows[i].Key < rows[j].Key }) } diff --git a/modules/mesh-delivery-forge-bridge/manifest_test.go b/modules/mesh-delivery-forge-bridge/manifest_test.go index 19686ac1..f702fa2f 100644 --- a/modules/mesh-delivery-forge-bridge/manifest_test.go +++ b/modules/mesh-delivery-forge-bridge/manifest_test.go @@ -28,12 +28,13 @@ type state struct { } type artifact struct { - Name string `json:"name"` - Kind string `json:"kind"` - Language string `json:"language"` - System string `json:"system"` - From string `json:"from"` - Binary string `json:"binary"` + Name string `json:"name"` + Kind string `json:"kind"` + Language string `json:"language"` + System string `json:"system"` + From string `json:"from"` + Binary string `json:"binary"` + Loads []string `json:"loads"` } type manifest struct { @@ -44,6 +45,7 @@ type manifest struct { Streams []stream `json:"streams"` State []state `json:"state"` Reads []string `json:"reads"` + Tools []string `json:"tools"` Invokes []string `json:"invokes"` OwnSecrets map[string]string `json:"own-secrets"` SecretsOwner string `json:"secrets-owner"` @@ -107,11 +109,15 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1 // §5), the revision of each key of the gate's buckets the bridge performed, so a key is performed once per // revision (ADR 0437 §8; issue 1341), and what a merge read of each module directory by its tree, so a merge reads // the forge only for the directories it changed (issue 1126). - if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}, {Name: "inputs", History: 1}}) { // 1279/AC6, 1341, 1126 - t.Errorf("state = %+v, want the buckets heads, performed and inputs", m.State) + if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}, {Name: "inputs", History: 1}, {Name: "end-asks", History: 1}}) { // 1279/AC6, 1341, 1126, 1585 + t.Errorf("state = %+v, want the buckets heads, performed, inputs and end-asks", m.State) } - // 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"} { + // 1585/T18: the tools read the groups and ask the bridge to say a group's end again, through its state end-asks. + if want := []string{"bridge_group_end_again", "bridge_groups"}; !reflect.DeepEqual(sorted(m.Tools), want) { + t.Errorf("tools = %v, want %v", m.Tools, want) + } + // No other state, no bus, no seat, no event (ADR 0437 §4; ADR 0521: no check.status). + for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the bridge has none", k) } @@ -143,8 +149,8 @@ func TestManifestProcess(t *testing.T) { // 1235/AC1 if _, bus := p["bus"]; bus { t.Errorf("the process declares a bus: it runs on its own broker credential") } - if len(m.Build.Artifacts) != 1 { - t.Fatalf("artifacts = %+v", m.Build.Artifacts) + if len(m.Build.Artifacts) != 2 || m.Build.Artifacts[1].Name != "tools" || m.Build.Artifacts[1].From != "cmd/mesh-delivery-forge-bridge-tools" { + t.Fatalf("artifacts = %+v, want the program and the tools bundle", m.Build.Artifacts) } a := m.Build.Artifacts[0] if st, err := os.Stat(a.From); err != nil || !st.IsDir() || a.Binary != "mesh-delivery-forge-bridge" || !strings.HasSuffix(a.From, a.Binary) { diff --git a/modules/mesh-delivery-forge-bridge/module.json b/modules/mesh-delivery-forge-bridge/module.json index f152cad0..338d9cdc 100644 --- a/modules/mesh-delivery-forge-bridge/module.json +++ b/modules/mesh-delivery-forge-bridge/module.json @@ -13,7 +13,12 @@ "state": [ { "name": "heads", "history": 1 }, { "name": "performed", "history": 1 }, - { "name": "inputs", "history": 1 } + { "name": "inputs", "history": 1 }, + { "name": "end-asks", "history": 1 } + ], + "tools": [ + "bridge_groups", + "bridge_group_end_again" ], "reads": [ "mesh-delivery-gate.merges", @@ -75,6 +80,17 @@ "system": "arch", "from": "cmd/mesh-delivery-forge-bridge", "binary": "mesh-delivery-forge-bridge" + }, + { + "name": "tools", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/mesh-delivery-forge-bridge-tools", + "binary": "mesh-delivery-forge-bridge-tools", + "loads": [ + "mesh-delivery-forge-bridge-tools" + ] } ] }