G-715: the forge bridge says a group's end once per base
ForgeGroupEnd gains base; the bridge reads each member's base from the seat's verb groups (Member.base is already in the contract and seatgen's output), keeps groups.<group>.<base>, says one end per base from that base's members, and bridge_group_end_again takes the base. Builds: design 47 G §715, A-236 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -101,11 +101,12 @@ func Tools(ask asker, now func() time.Time, groupsRead func() ([]groups.Row, []s
|
||||
"writes it again from the seat mesh-delivery's groups, 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's merged message is not written. Answers whether it was written.",
|
||||
Input: map[string]any{"type": "object", "required": []string{"group"}, "properties": map[string]any{
|
||||
Input: map[string]any{"type": "object", "required": []string{"group", "base"}, "properties": map[string]any{
|
||||
"group": map[string]any{"type": "string", "description": "the group, by its id (its branch)"},
|
||||
"base": map[string]any{"type": "string", "description": "the watched branch its members merged into, whose end is said again"},
|
||||
"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())
|
||||
key, value, err := groups.NewEndAsk(str(args, "group"), str(args, "base"), str(args, "why"), stdio.CallerOf(args), now())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -113,7 +114,7 @@ func Tools(ask asker, now func() time.Time, groupsRead func() ([]groups.Row, []s
|
||||
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
|
||||
return map[string]any{"group": a.Group, "base": a.Base, "written": a.Written, "answer": a.Answer, "by": a.By, "why": a.Why}, nil
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
+17
-8
@@ -27,8 +27,8 @@ 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)
|
||||
grp := func(id string, a, b int, stateB string) seat.Group {
|
||||
return seat.Group{ID: id, State: "merging", Members: []seat.Member{{Repository: "novox/a", Number: int64(a), Head: h1, State: "merged"},
|
||||
{Repository: "novox/b", Number: int64(b), Head: h2, State: stateB}}}
|
||||
return seat.Group{ID: id, State: "merging", Members: []seat.Member{{Repository: "novox/a", Number: int64(a), Head: h1, State: "merged", Base: "main"},
|
||||
{Repository: "novox/b", Number: int64(b), Head: h2, State: stateB, Base: "main"}}}
|
||||
}
|
||||
of := []seat.Group{grp("ended", 1, 2, "merged"), grp("ready", 3, 4, "merged"), grp("gathered", 5, 6, "merging"),
|
||||
{ID: "solo", Members: []seat.Member{{Repository: "novox/a", Number: 7, Head: h1}}}, grp("", 8, 9, "merged")}
|
||||
@@ -40,9 +40,9 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) {
|
||||
published[groups.PublishedKey("novox/b", n)] = c2
|
||||
}
|
||||
// The state end-asks keeps the last ask of the group ready, answered, and one that does not read.
|
||||
answeredAsk, _ := json.Marshal(groups.EndAsk{V: 1, Group: "ready", Why: "lost", By: "lead@shanks", At: time.Date(2026, 11, 4, 22, 0, 0, 0, time.UTC),
|
||||
answeredAsk, _ := json.Marshal(groups.EndAsk{V: 1, Group: "ready", Base: "main", Why: "lost", By: "lead@shanks", At: time.Date(2026, 11, 4, 22, 0, 0, 0, time.UTC),
|
||||
Written: true, Answer: "the end of the group is written again", Answered: time.Date(2026, 11, 4, 22, 0, 1, 0, time.UTC)})
|
||||
asks := map[string][]byte{"ready": answeredAsk, "gathered": []byte("{")}
|
||||
asks := map[string][]byte{"ready.main": answeredAsk, "gathered.main": []byte("{")}
|
||||
var asked groups.EndAsk
|
||||
var askedKey string
|
||||
ask := func(key string, value []byte) (groups.EndAsk, error) {
|
||||
@@ -54,7 +54,7 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) {
|
||||
}
|
||||
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(of, published, map[string]string{"groups.ended": c2}, asks)
|
||||
rows, unread := groups.Rows(of, published, map[string]string{"groups.ended.main": c2}, asks)
|
||||
return rows, unread, nil
|
||||
}, nil, nil)
|
||||
call := func(name string, args map[string]any) any {
|
||||
@@ -83,7 +83,7 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) {
|
||||
for _, r := range out.Groups {
|
||||
m[r.Group] = r
|
||||
}
|
||||
if len(args) == 0 && strings.Join(out.Unread, ", ") != "a group with no id, end-asks.gathered" {
|
||||
if len(args) == 0 && strings.Join(out.Unread, ", ") != "a group with no id, end-asks.gathered.main" {
|
||||
t.Errorf("unread = %v, want the group with no id and end-asks.gathered", out.Unread)
|
||||
}
|
||||
return m
|
||||
@@ -107,8 +107,8 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) {
|
||||
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 {
|
||||
out := call(EndAgainTool, map[string]any{"group": "ready", "base": "main", "why": "the indexer never took it in"}).(map[string]any)
|
||||
if askedKey != "ready.main" || asked.Group != "ready" || asked.Base != "main" || 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 {
|
||||
@@ -116,9 +116,18 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) {
|
||||
if _, err := tl.Run(map[string]any{}); err == nil {
|
||||
t.Error("an ask naming no group was taken")
|
||||
}
|
||||
if _, err := tl.Run(map[string]any{"group": "ready"}); err == nil {
|
||||
t.Error("an ask naming no base was taken")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// §715: a group with members on two bases is two rows, one per base, each with its own members and key.
|
||||
split := seat.Group{ID: "feat/s", State: "merging", Members: []seat.Member{{Repository: "novox/a", Number: 11, Base: "beta"}, {Repository: "novox/b", Number: 12, Base: "main"}, {Repository: "novox/c", Number: 13, Base: "beta"}}}
|
||||
if rows, unread := groups.Rows([]seat.Group{split}, nil, nil, nil); len(unread) != 0 || len(rows) != 2 || rows[0].Base != "beta" || len(rows[0].Members) != 2 || rows[1].Base != "main" || len(rows[1].Members) != 1 || rows[0].Key == rows[1].Key {
|
||||
t.Errorf("rows of a group on two bases = %+v unread %v", rows, unread)
|
||||
}
|
||||
|
||||
// 1585: bridge_merged_message answers the merged message as the bridge publishes it: for a lone pull request the same bytes
|
||||
// the event gives through Convert, for a member of a delivery group those with its group and members set as the process
|
||||
// sets them; a merge the gate does not hold, or a forge that does not answer, is said.
|
||||
|
||||
@@ -58,10 +58,10 @@ func (g *grouper) answerEnd(ctx context.Context, a groups.EndAsk, at time.Time)
|
||||
switch {
|
||||
case !ok || len(m.Members) < 2:
|
||||
a.Answer = "the seat " + seat.Seat + " holds no group " + a.Group + " of more than one member"
|
||||
case g.end(ctx, m, groups.AgainStamp(a.At)):
|
||||
case g.end(ctx, m, a.Base, 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's merged message is not written, or FORGE refused it (the bridge's journal says which)"
|
||||
a.Answer = "the end on " + a.Base + " is not written: a member's merged message is not written, or FORGE refused it (the bridge's journal says which)"
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
@@ -20,7 +20,8 @@ import (
|
||||
// What it keeps of it is in its state performed: `published.<hash>` for each member's merged message written, its value
|
||||
// the merge commit it named (the hash is the first 16 hex digits of the sha256 of `<repository>#<number>`), and
|
||||
// `groups.<group>` once the group's end is written, its value the last member's merge commit. A group whose end is not
|
||||
// written yet is open (bridge_groups reads them).
|
||||
// written yet is open (bridge_groups reads them). A group gathers per watched branch: its end is said once per base, from
|
||||
// the members whose base the seat says is that branch, and its key is `groups.<group>.<base>` (design 47 G §715).
|
||||
|
||||
// groupsRead is every delivery group the seat answers.
|
||||
type groupsRead func(ctx context.Context) ([]seat.Group, error)
|
||||
@@ -93,19 +94,24 @@ func (g *grouper) published(ctx context.Context, f forgemsg.Forge) {
|
||||
return
|
||||
}
|
||||
if m, ok := g.groupOf(ctx, f); ok {
|
||||
g.end(ctx, m, "")
|
||||
g.end(ctx, m, f.Base, "")
|
||||
}
|
||||
}
|
||||
|
||||
// end says the group's end (forge.group.merged) once every member's merged message is written, after the last of them;
|
||||
// never on a refusal (ADR 0441 §3: a held group merges on from it), since a refused member's message is never written. It
|
||||
// is written once per group: `groups.<group>` in performed says it was. again says it is written again by hand
|
||||
// is written once per group and base: `groups.<group>.<base>` in performed says it was. again says it is written again by hand
|
||||
// (bridge_group_end_again), with the time appended to the message id, so the duplicate window does not drop it.
|
||||
func (g *grouper) end(ctx context.Context, m seat.Group, again string) bool {
|
||||
func (g *grouper) end(ctx context.Context, m seat.Group, base, again string) bool {
|
||||
if len(m.Members) < 2 {
|
||||
return false
|
||||
}
|
||||
key, err := groupkeys.Key(m.ID)
|
||||
onBase := groupkeys.OnBase(m, base)
|
||||
if len(onBase) == 0 {
|
||||
g.logger.Printf("the group %q has no member on %q, so no end is said on it", m.ID, base)
|
||||
return false
|
||||
}
|
||||
key, err := groupkeys.Key(m.ID, base)
|
||||
if err != nil {
|
||||
g.logger.Printf("the group %q cannot be kept, so its end is not said: %v", m.ID, err)
|
||||
return false
|
||||
@@ -116,8 +122,8 @@ func (g *grouper) end(ctx context.Context, m seat.Group, again string) bool {
|
||||
}
|
||||
}
|
||||
// The members in the group's order, which the gate merges them in (ADR 0441 §5), so the last commit names the end.
|
||||
end := forgemsg.ForgeGroupEnd{Group: m.ID}
|
||||
for _, member := range m.Members {
|
||||
end := forgemsg.ForgeGroupEnd{Group: m.ID, Base: base}
|
||||
for _, member := range onBase {
|
||||
merge, ok, err := g.done.Get(ctx, groupkeys.PublishedKey(member.Repository, int(member.Number)))
|
||||
if err != nil || !ok || merge == "" {
|
||||
return false // its merged message is not written yet: said when it is (published)
|
||||
@@ -130,13 +136,13 @@ func (g *grouper) end(ctx context.Context, m seat.Group, again string) bool {
|
||||
return false
|
||||
}
|
||||
last := end.Members[len(end.Members)-1].Merge
|
||||
if err := g.send(ctx, forgeSubject(forgemsg.ForgeGroupEndEvent), data, groupkeys.EndID(m.ID, last, again)); err != nil {
|
||||
if err := g.send(ctx, forgeSubject(forgemsg.ForgeGroupEndEvent), data, groupkeys.EndID(m.ID, base, last, again)); err != nil {
|
||||
g.logger.Printf("the end of the group %q is not written on FORGE, written by bridge_group_end_again: %v", m.ID, err)
|
||||
return false
|
||||
}
|
||||
if err := g.done.Put(ctx, key, last); err != nil {
|
||||
g.logger.Printf("the end of the group %q is written and not kept: the stream's duplicate window drops a second one: %v", m.ID, err)
|
||||
}
|
||||
g.logger.Printf("the end of the group %q is written: %d members merged, the last %s", m.ID, len(end.Members), last)
|
||||
g.logger.Printf("the end of the group %q on %q is written: %d members merged, the last %s", m.ID, base, len(end.Members), last)
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -33,7 +33,8 @@ func TestAGroupsMembersNameItAndItsEndIsSaidOnce(t *testing.T) {
|
||||
h1, h2 := strings.Repeat("1", 40), strings.Repeat("2", 40)
|
||||
c1, c2 := strings.Repeat("c", 40), strings.Repeat("d", 40)
|
||||
group := seat.Group{ID: "feat/take-in", State: "merging", Members: []seat.Member{
|
||||
{Repository: "novox/mesh-sdk", Number: 37, Head: h1, State: "merged"}, {Repository: "novox/mesh-delivery", Number: 9, Head: h2, State: "merging"}}}
|
||||
{Repository: "novox/mesh-sdk", Number: 37, Head: h1, State: "merged", Base: "beta"}, {Repository: "novox/mesh-delivery", Number: 9, Head: h2, State: "merging", Base: "beta"},
|
||||
{Repository: "novox/mesh-tools", Number: 5, Head: h2, State: "merging", Base: "main"}}}
|
||||
var of []seat.Group
|
||||
var read error
|
||||
var logs bytes.Buffer
|
||||
@@ -41,7 +42,7 @@ func TestAGroupsMembersNameItAndItsEndIsSaidOnce(t *testing.T) {
|
||||
g := &grouper{send: pub.send, done: done, logger: log.New(&logs, "", 0),
|
||||
groups: func(context.Context) ([]seat.Group, error) { return of, read }}
|
||||
mergedOf := func(repo string, number int, head, commit string) []byte {
|
||||
data, err := forgemsg.EncodeForge(forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: repo, Branch: "feat/take-in", Number: number, Head: head, Merge: commit, Root: "r"})
|
||||
data, err := forgemsg.EncodeForge(forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: repo, Branch: "feat/take-in", Base: "beta", Number: number, Head: head, Merge: commit, Root: "r"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -60,7 +61,7 @@ func TestAGroupsMembersNameItAndItsEndIsSaidOnce(t *testing.T) {
|
||||
read = nil
|
||||
// The first member: its message names the group and both members in the group's order; the group is not ended.
|
||||
data, f, ok := g.member(ctx, mergedOf("novox/mesh-sdk", 37, h1, c1))
|
||||
if back, err := forgemsg.DecodeForge(data); !ok || err != nil || back.Group != "feat/take-in" || len(back.Members) != 2 || back.Members[0].Number != 37 {
|
||||
if back, err := forgemsg.DecodeForge(data); !ok || err != nil || back.Group != "feat/take-in" || len(back.Members) != 3 || back.Members[0].Number != 37 {
|
||||
t.Fatalf("a member's message: %+v %v", back, err)
|
||||
}
|
||||
g.published(ctx, f)
|
||||
@@ -71,20 +72,27 @@ func TestAGroupsMembersNameItAndItsEndIsSaidOnce(t *testing.T) {
|
||||
_, f, _ = g.member(ctx, mergedOf("novox/mesh-delivery", 9, h2, c2))
|
||||
g.published(ctx, f)
|
||||
g.published(ctx, f)
|
||||
if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.group.merged" || pub.ids[0] != "group-end.feat/take-in."+c2 {
|
||||
t.Fatalf("ends %v ids %v, want one forge.group.merged with id group-end.feat/take-in.%s", pub.subjects, pub.ids, c2)
|
||||
if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.group.merged" || pub.ids[0] != "group-end.feat/take-in.beta."+c2 {
|
||||
t.Fatalf("ends %v ids %v, want one forge.group.merged with id group-end.feat/take-in.beta.%s", pub.subjects, pub.ids, c2)
|
||||
}
|
||||
end, err := forgemsg.DecodeForgeGroupEnd(pub.data[0])
|
||||
if err != nil || len(end.Members) != 2 || end.Members[0].Merge != c1 || end.Members[1].Merge != c2 {
|
||||
t.Errorf("the end = %+v %v, want both members in the group's order", end, err)
|
||||
if err != nil || end.Base != "beta" || len(end.Members) != 2 || end.Members[0].Merge != c1 || end.Members[1].Merge != c2 {
|
||||
t.Errorf("the end = %+v %v, want the two members on beta in the group's order", end, err)
|
||||
}
|
||||
if key, _ := groupkeys.Key("feat/take-in"); done[key] != c2 {
|
||||
if key, _ := groupkeys.Key("feat/take-in", "beta"); done[key] != c2 {
|
||||
t.Errorf("performed %v: want groups.feat~take-in kept with the last merge", done)
|
||||
}
|
||||
// §715: the member on main is on another base, and its merged message ends the group on main alone, once.
|
||||
done[groupkeys.PublishedKey("novox/mesh-tools", 5)] = c1
|
||||
g.published(ctx, forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: "novox/mesh-tools", Branch: "feat/take-in", Base: "main", Number: 5, Merge: c1})
|
||||
g.published(ctx, forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: "novox/mesh-tools", Branch: "feat/take-in", Base: "main", Number: 5, Merge: c1})
|
||||
if mainEnd, err := forgemsg.DecodeForgeGroupEnd(pub.data[len(pub.data)-1]); len(pub.subjects) != 2 || err != nil || mainEnd.Base != "main" || len(mainEnd.Members) != 1 || mainEnd.Members[0].Number != 5 {
|
||||
t.Errorf("ends %d, the last %+v %v: want one more end, on main, of the member on main alone", len(pub.subjects), mainEnd, err)
|
||||
}
|
||||
// 1585/T19: an ask to say a group's end again is answered from the seat's groups, with the ask's time on the message id;
|
||||
// not written while a member's merged message is not written, with the reason; not for a group the seat does not hold,
|
||||
// nor when the groups cannot be read.
|
||||
ask := groupkeys.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)}
|
||||
ask := groupkeys.EndAsk{V: 1, Group: "feat/take-in", Base: "beta", 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
|
||||
of []seat.Group
|
||||
@@ -94,6 +102,7 @@ func TestAGroupsMembersNameItAndItsEndIsSaidOnce(t *testing.T) {
|
||||
answer string
|
||||
}{
|
||||
{"every message written: written again", []seat.Group{group}, nil, true, true, "written again"},
|
||||
{"a base no member is on", []seat.Group{{ID: group.ID, Members: []seat.Member{group.Members[2], {Repository: "novox/mesh-sdk", Number: 37, Base: "main"}}}}, nil, true, false, "not written"},
|
||||
{"a member's message not written", []seat.Group{group}, nil, false, false, "not written"},
|
||||
{"no such group at the seat", []seat.Group{{ID: "other", Members: group.Members}}, nil, true, false, "holds no group"},
|
||||
{"the groups cannot be read", nil, errors.New("no answer"), true, false, "cannot be read"},
|
||||
@@ -101,6 +110,7 @@ func TestAGroupsMembersNameItAndItsEndIsSaidOnce(t *testing.T) {
|
||||
pub, done := &idPub{}, memStrs{}
|
||||
if tc.published {
|
||||
done[groupkeys.PublishedKey("novox/mesh-sdk", 37)], done[groupkeys.PublishedKey("novox/mesh-delivery", 9)] = c1, c2
|
||||
done[groupkeys.PublishedKey("novox/mesh-tools", 5)] = c1
|
||||
}
|
||||
g := &grouper{send: pub.send, done: done, logger: log.New(&bytes.Buffer{}, "", 0),
|
||||
groups: func(context.Context) ([]seat.Group, error) { return tc.of, tc.read }}
|
||||
@@ -108,7 +118,7 @@ func TestAGroupsMembersNameItAndItsEndIsSaidOnce(t *testing.T) {
|
||||
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] != groupkeys.EndID("feat/take-in", c2, groupkeys.AgainStamp(ask.At)) {
|
||||
if tc.written && pub.ids[0] != groupkeys.EndID("feat/take-in", "beta", c2, groupkeys.AgainStamp(ask.At)) {
|
||||
t.Errorf("%s: the id %s does not carry the ask's time", tc.name, pub.ids[0])
|
||||
}
|
||||
}
|
||||
|
||||
+2
-2
@@ -754,7 +754,7 @@ func TestBridgeSaysAGroupFromTheSeatsGroups(t *testing.T) { // 1585, ADR 0633 §
|
||||
headB := strings.Repeat("b", 40)
|
||||
mergeB := strings.Repeat("e", 40)
|
||||
groups, _ := json.Marshal(seat.GroupsAnswer{Groups: []seat.Group{{ID: "feat/x", State: "merging", Members: []seat.Member{
|
||||
{Repository: "novox/mesh-delivery", Number: 12, Head: head, State: "merged"}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB, State: "merging"}}}}})
|
||||
{Repository: "novox/mesh-delivery", Number: 12, Head: head, State: "merged", Base: "main"}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB, State: "merging", Base: "main"}}}}})
|
||||
b.answer("mesh-delivery.groups", string(groups))
|
||||
ch := heard(t, b)
|
||||
landedOn(b, "novox/mesh-delivery", 12, "feat/x", head, merge)
|
||||
@@ -768,7 +768,7 @@ func TestBridgeSaysAGroupFromTheSeatsGroups(t *testing.T) { // 1585, ADR 0633 §
|
||||
}
|
||||
}
|
||||
end, err := forgemsg.DecodeForgeGroupEnd(next(t, ch, forgeSubject(forgemsg.ForgeGroupEndEvent)))
|
||||
if err != nil || end.Group != "feat/x" || len(end.Members) != 2 || end.Members[0].Merge != merge || end.Members[1].Merge != mergeB {
|
||||
if err != nil || end.Group != "feat/x" || end.Base != "main" || len(end.Members) != 2 || end.Members[0].Merge != merge || end.Members[1].Merge != mergeB {
|
||||
t.Errorf("the group's end: %+v %v", end, err)
|
||||
}
|
||||
refusedRead(t, b, "mesh-delivery-gate_merges")
|
||||
|
||||
@@ -19,11 +19,13 @@ type GroupMerge struct {
|
||||
Merge string `json:"merge"`
|
||||
}
|
||||
|
||||
// ForgeGroupEnd is the body of ForgeGroupEndEvent: the group and every member merged in it, with its merge commit.
|
||||
// ForgeGroupEnd is the body of ForgeGroupEndEvent: the group and every member merged in it on its base, with its merge commit.
|
||||
// The indexer takes the members it holds in as one source revision once it hears it (ADR 0633 §4).
|
||||
type ForgeGroupEnd struct {
|
||||
V int `json:"v"`
|
||||
Group string `json:"group"`
|
||||
V int `json:"v"`
|
||||
Group string `json:"group"`
|
||||
// Base is the watched branch the members merged into: a group's end is said once per base (design 47 G §715).
|
||||
Base string `json:"base,omitempty"`
|
||||
Members []GroupMerge `json:"members"`
|
||||
}
|
||||
|
||||
|
||||
@@ -34,22 +34,51 @@ const (
|
||||
EndAsksBucket = Module + "_" + EndAsksState
|
||||
)
|
||||
|
||||
// Key is the key of performed that says a group's end is written: `groups.<group, escaped as a key>`, its value the last
|
||||
// member's merge commit the end named.
|
||||
func Key(group string) (string, error) {
|
||||
part, err := KeyOf(group)
|
||||
// Key is the key of performed that says a group's end on a base is written: `groups.<group>.<base>`, each part escaped as
|
||||
// a key name, its value the last member's merge commit the end named (design 47 G §715: one end per base).
|
||||
func Key(group, base string) (string, error) {
|
||||
part, err := KeyOf(group, base)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return "groups." + part, nil
|
||||
}
|
||||
|
||||
// KeyOf is a group's id (its branch) escaped as a key, as the states performed and end-asks key it.
|
||||
func KeyOf(group string) (string, error) {
|
||||
// KeyOf is a group's id (its branch) and a base, each escaped as a key name, as the states performed and end-asks key
|
||||
// them: `<group>.<base>`.
|
||||
func KeyOf(group, base string) (string, error) {
|
||||
if strings.TrimSpace(group) == "" {
|
||||
return "", errors.New("a group's id is empty")
|
||||
}
|
||||
return keys.EscapeName(group), nil
|
||||
if strings.TrimSpace(base) == "" {
|
||||
return "", errors.New("a group's base is empty")
|
||||
}
|
||||
return keys.EscapeName(group) + "." + keys.EscapeName(base), nil
|
||||
}
|
||||
|
||||
// Bases are the distinct bases of a group's members, in the order the members name them; a member naming none is
|
||||
// named in missing by its repository and number.
|
||||
func Bases(g seat.Group) (bases, missing []string) {
|
||||
for _, m := range g.Members {
|
||||
switch {
|
||||
case m.Base == "":
|
||||
missing = append(missing, fmt.Sprintf("%s#%d", m.Repository, m.Number))
|
||||
case !slices.Contains(bases, m.Base):
|
||||
bases = append(bases, m.Base)
|
||||
}
|
||||
}
|
||||
return bases, missing
|
||||
}
|
||||
|
||||
// OnBase is the members of a group whose base is base, in the group's order.
|
||||
func OnBase(g seat.Group, base string) []seat.Member {
|
||||
var out []seat.Member
|
||||
for _, m := range g.Members {
|
||||
if m.Base == base {
|
||||
out = append(out, m)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// PublishedKey is the key of performed that says a member's merged message is written: `published.<16 hex digits of the
|
||||
@@ -59,10 +88,10 @@ func PublishedKey(repository string, number int) string {
|
||||
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
|
||||
// EndID is the message id of a group's end on a base, 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, base, lastMerge, again string) string {
|
||||
id := "group-end." + group + "." + base + "." + lastMerge
|
||||
if again != "" {
|
||||
id += "." + again
|
||||
}
|
||||
@@ -77,6 +106,7 @@ func AgainStamp(at time.Time) string { return at.UTC().Format("20060102T150405.0
|
||||
type EndAsk struct {
|
||||
V int `json:"v"`
|
||||
Group string `json:"group"`
|
||||
Base string `json:"base"`
|
||||
Why string `json:"why,omitempty"`
|
||||
By string `json:"by,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
@@ -89,16 +119,19 @@ type EndAsk struct {
|
||||
func (a EndAsk) Done() bool { return !a.Answered.IsZero() }
|
||||
|
||||
// NewEndAsk is the value bridge_group_end_again writes for a group: its key (the group escaped as a key) and the value.
|
||||
func NewEndAsk(group, why, by string, at time.Time) (string, []byte, error) {
|
||||
group = strings.TrimSpace(group)
|
||||
func NewEndAsk(group, base, why, by string, at time.Time) (string, []byte, error) {
|
||||
group, base = strings.TrimSpace(group), strings.TrimSpace(base)
|
||||
if group == "" {
|
||||
return "", nil, errors.New("name the group whose end is said again (group)")
|
||||
}
|
||||
key, err := KeyOf(group)
|
||||
if base == "" {
|
||||
return "", nil, errors.New("name the watched branch its end is said again on (base)")
|
||||
}
|
||||
key, err := KeyOf(group, base)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
data, err := json.Marshal(EndAsk{V: 1, Group: group, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
|
||||
data, err := json.Marshal(EndAsk{V: 1, Group: group, Base: base, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
|
||||
return key, data, err
|
||||
}
|
||||
|
||||
@@ -108,16 +141,17 @@ func DecodeEndAsk(data []byte) (EndAsk, error) {
|
||||
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")
|
||||
if a.V != 1 || a.Group == "" || a.Base == "" {
|
||||
return a, errors.New("an end-asks value of no version 1 naming no group and base")
|
||||
}
|
||||
return a, nil
|
||||
}
|
||||
|
||||
// Row is one group as bridge_groups answers it: the seat mesh-delivery's group and what the bridge kept of its end.
|
||||
// Row is one group on one base as bridge_groups answers it: the seat mesh-delivery's group and what the bridge kept of its end.
|
||||
type Row struct {
|
||||
Key string `json:"key"`
|
||||
Group string `json:"group"`
|
||||
Base string `json:"base"`
|
||||
State string `json:"state,omitempty"`
|
||||
Members []RowMember `json:"members"`
|
||||
Ended bool `json:"ended"`
|
||||
@@ -159,30 +193,36 @@ func Rows(of []seat.Group, published, ended map[string]string, asks map[string][
|
||||
if len(g.Members) < 2 {
|
||||
continue
|
||||
}
|
||||
key, err := KeyOf(g.ID)
|
||||
if err != nil {
|
||||
unread = append(unread, "a group with no id")
|
||||
continue
|
||||
bases, missing := Bases(g)
|
||||
for _, m := range missing {
|
||||
unread = append(unread, "the member "+m+" of the group "+g.ID+" names no base")
|
||||
}
|
||||
r := Row{Key: key, Group: g.ID, State: g.State}
|
||||
allPublished := true
|
||||
for _, m := range g.Members {
|
||||
commit, pub := published[PublishedKey(m.Repository, int(m.Number))]
|
||||
r.Members = append(r.Members, RowMember{Repository: m.Repository, Number: int(m.Number), State: m.State, Commit: commit, Published: pub})
|
||||
allPublished = allPublished && pub
|
||||
for _, base := range bases {
|
||||
key, err := KeyOf(g.ID, base)
|
||||
if err != nil {
|
||||
unread = append(unread, "a group with no id")
|
||||
continue
|
||||
}
|
||||
r := Row{Key: key, Group: g.ID, Base: base, State: g.State}
|
||||
allPublished := true
|
||||
for _, m := range OnBase(g, base) {
|
||||
commit, pub := published[PublishedKey(m.Repository, int(m.Number))]
|
||||
r.Members = append(r.Members, RowMember{Repository: m.Repository, Number: int(m.Number), State: m.State, Commit: commit, Published: pub})
|
||||
allPublished = allPublished && pub
|
||||
}
|
||||
r.EndedWith, r.Ended = ended["groups."+key]
|
||||
r.LastEndAsk = lastAsk[key]
|
||||
r.Open = !r.Ended
|
||||
switch {
|
||||
case r.Ended:
|
||||
r.Words = "its end is written"
|
||||
case allPublished:
|
||||
r.Words = "every member's merged message is written, and its end is not written: bridge_group_end_again says it"
|
||||
default:
|
||||
r.Words = "gathered: not every member's merged message is written"
|
||||
}
|
||||
rows = append(rows, r)
|
||||
}
|
||||
r.EndedWith, r.Ended = ended["groups."+key]
|
||||
r.LastEndAsk = lastAsk[key]
|
||||
r.Open = !r.Ended
|
||||
switch {
|
||||
case r.Ended:
|
||||
r.Words = "its end is written"
|
||||
case allPublished:
|
||||
r.Words = "every member's merged message is written, and its end is not written: bridge_group_end_again says it"
|
||||
default:
|
||||
r.Words = "gathered: not every member's merged message is written"
|
||||
}
|
||||
rows = append(rows, r)
|
||||
}
|
||||
sortRows(rows)
|
||||
sort.Strings(unread)
|
||||
|
||||
Reference in New Issue
Block a user