Forge bridge: bridge_groups and bridge_group_end_again, the end said again through the state end-asks (issue 1585, ADR 0633 §4)
Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
This commit is contained in:
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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.<merges key>`, 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 <repository>#<number>@<merge>>`.
|
||||
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 }) }
|
||||
@@ -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) {
|
||||
|
||||
@@ -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"
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user