novox/mesh-sdk
18
0
Fork 0
Code Issues 1 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

delivery: the take-in events, their keys and body (ADR 0633 §1, §5, issue 1585)

Browse Source 
The indexer says a wait for an indexer build or a group's end with
take-in.pending and its end with take-in.taken, on its own stream, so the
watchman can bound the wait and a hand act can say what it ended.

Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
This commit is contained in:
jschoubben
2026-11-04 22:51:47 +01:00
parent dea2a022bd
commit 8ba75aae89
2 changed files with 240 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/delivery/source_test.go
+72 -1
View File
@@ -59,11 +59,29 @@ func TestSourceKeys(t *testing.T) { // 1200/AC1
t.Errorf("error does not name the kind: %v", err)
}
}
for _, bad := range []string{"", "forge.a.b", "forge.a.b.closed", "x.a.b.pull", "forge.a.b.pull.x", "forge..b.pull"} {
for _, bad := range []string{"", "forge.a.b", "forge.a.b.closed", "x.a.b.pull", "forge.a.b.pull.x", "forge..b.pull", ForgeGroupEndEvent} {
if _, _, err := ParseForgeEvent(bad); err == nil {
t.Errorf("parse forge event %q accepted", bad)
}
}
// 1585/T2: the take-in events, `take-in.<pending|taken>.<key>`, a group's key with its dots and no subject wildcard.
if PendingKey(42) != "m.00000000000000000042" || GroupTakeInKey("feat/a.b *>") != "g.feat/a.b___" {
t.Errorf("take-in keys %q %q", PendingKey(42), GroupTakeInKey("feat/a.b *>"))
}
for _, c := range []struct{ kind, key string }{{TakeInPending, PendingKey(7)}, {TakeInTaken, GroupTakeInKey("feat/a.b")}} {
ev, err := TakeInEvent(c.kind, c.key)
if err != nil || ev != "take-in."+c.kind+"."+c.key {
t.Fatalf("take-in event = %q, %v", ev, err)
}
if k, key, err := ParseTakeInEvent(ev); err != nil || k != c.kind || key != c.key {
t.Errorf("parse take-in event %q = %q %q %v", ev, k, key, err)
}
}
for _, bad := range []string{"", "take-in.pending", "take-in.held.m.1", "take-in.taken.x.1", "take-in.taken.m..1", "forge.taken.m.1"} {
if _, _, err := ParseTakeInEvent(bad); err == nil {
t.Errorf("parse take-in event %q accepted", bad)
}
}
}
func TestForgeOpenKind(t *testing.T) { // 1279/AC5
@@ -156,6 +174,18 @@ func sampleForges() []Forge {
}
}
// sampleTakeIns are a pending revision, a group being gathered and a hand take-in's taken (1585/T2).
func sampleTakeIns(since time.Time) []TakeIn {
m := []GroupMerge{{Repository: "novox/mesh-delivery", Number: 9, Merge: "eeee"}}
return []TakeIn{
{Kind: TakeInPending, Key: PendingKey(12), What: TakeInRevision, Merges: m, WaitsFor: "src2:abc", Since: since},
{Kind: TakeInPending, Key: GroupTakeInKey("feat/take-in"), What: TakeInGroup, Merges: m, Group: "feat/take-in",
Members: []Head{{Repository: "novox/mesh-delivery", Number: 9, Head: "ffff"}, {Repository: "novox/mesh-sdk", Number: 37, Head: "dddd"}}, Since: since},
{Kind: TakeInTaken, Key: PendingKey(12), What: TakeInRevision, Merges: m, WaitsFor: "src2:abc", Since: since,
By: "lead-47@shanks", Why: "the named indexer can never run"},
}
}
func TestSourceValuesRoundTrip(t *testing.T) { // 1200/AC3
check := func(name string, data []byte, err error) {
t.Helper()
@@ -210,6 +240,16 @@ func TestSourceValuesRoundTrip(t *testing.T) { // 1200/AC3
if got, err := DecodeForgeGroupEnd(data); err != nil || !SameForgeGroupEnd(got, ge) {
t.Errorf("group end round trip: %+v %v", got, err)
}
// 1585/T2: the take-in bodies: a pending revision, a group being gathered, and a hand take-in's taken with its by.
since := time.Date(2026, 11, 4, 23, 0, 0, 0, time.UTC)
for _, ti := range sampleTakeIns(since) {
data, err = EncodeTakeIn(ti)
check("take-in "+ti.Kind+" "+ti.What, data, err)
ti.V = Version
if got, err := DecodeTakeIn(data); err != nil || !SameTakeIn(got, ti) {
t.Errorf("take-in %s round trip: %+v %v", ti.Key, got, err)
}
}
// 1280: the taken and reach values (ADR 0556).
tk := Taken{Repository: "novox/mesh-sdk", Merge: "cccc", At: time.Date(2026, 10, 26, 20, 0, 0, 0, time.UTC)}
data, err = EncodeTaken(tk)
@@ -232,6 +272,7 @@ func TestSourceValuesRoundTrip(t *testing.T) { // 1200/AC3
"module": func(b []byte) error { _, err := DecodeModule(b); return err },
"forge": func(b []byte) error { _, err := DecodeForge(b); return err },
"group end": func(b []byte) error { _, err := DecodeForgeGroupEnd(b); return err },
"take-in": func(b []byte) error { _, err := DecodeTakeIn(b); return err },
} {
for _, in := range []string{`{"group":"g"}`, `{"v":2,"group":"g"}`} {
if err := dec([]byte(in)); !errors.Is(err, ErrUnknownVersion) {
@@ -345,6 +386,36 @@ func TestSourceValuesRefused(t *testing.T) { // 1200/AC4
wantErr("group end without a group", err, "names its group")
_, err = EncodeForgeGroupEnd(ForgeGroupEnd{Group: "g", Members: []GroupMerge{{Repository: "a/b", Number: 1}}})
wantErr("group end with a member without its merge", err, "merge commit")
// 1585/T2: a take-in body that does not hold.
since := time.Date(2026, 11, 4, 23, 0, 0, 0, time.UTC)
ti := sampleTakeIns(since)[1]
ti.Group = ""
_, err = EncodeTakeIn(ti)
wantErr("group take-in without its group", err, "names its group")
ti = sampleTakeIns(since)[1]
ti.Key = GroupTakeInKey("other")
_, err = EncodeTakeIn(ti)
wantErr("group take-in under another group's key", err, "g.other")
ti = sampleTakeIns(since)[0]
ti.WaitsFor = ""
_, err = EncodeTakeIn(ti)
wantErr("pending revision waiting for no build", err, "waits for")
ti = sampleTakeIns(since)[0]
ti.By = "someone@node"
_, err = EncodeTakeIn(ti)
wantErr("pending with a hand act", err, "only a taken event")
ti = sampleTakeIns(since)[2]
ti.By = ""
_, err = EncodeTakeIn(ti)
wantErr("hand act without its by", err, "by")
ti = sampleTakeIns(since)[0]
ti.Merges = nil
_, err = EncodeTakeIn(ti)
wantErr("take-in without merges", err, "no merge")
ti = sampleTakeIns(since)[0]
ti.Since = time.Time{}
_, err = EncodeTakeIn(ti)
wantErr("take-in without since", err, "since")
_, err = EncodeForge(Forge{Kind: "closed", Repository: "a/b"})
wantErr("unknown kind", err, "closed")
_, err = EncodeForge(Forge{Kind: ForgePull, Repository: "nope", Number: 1, Head: "h", Action: "opened"})
go/delivery/takein.go
+168
View File
@@ -0,0 +1,168 @@
package delivery
import (
"encoding/json"
"fmt"
"reflect"
"strings"
"time"
)
// The take-in events (novox/hq ADR 0633 §1, §5): the indexer publishes them on its own stream, which keeps the last
// message per subject, `take-in.<kind>.<key>`. A wait starts with `take-in.pending` and ends with `take-in.taken`,
// whoever ended it (the named indexer, a revert, a hand take-in or a drop). The stream's name is the indexer's manifest's;
// these names are its events, relative to it.
const (
TakeInStream = "take-in"
TakeInPending = "pending"
TakeInTaken = "taken"
)
// What waits in a take-in event: a source revision for the indexer build it names, or a delivery group for its end.
const (
TakeInRevision = "revision"
TakeInGroup = "group"
)
// PendingKey is the key of one held merge in the indexer's pending state, and the key a revision's take-in events
// carry (its oldest merge's): `m.<the merge's stream sequence, 20 digits>`, so the key order is the merge order.
func PendingKey(seq uint64) string { return fmt.Sprintf("m.%020d", seq) }
// GroupTakeInKey is the key a group's take-in events carry while it is gathered: `g.<group>`, with every character a
// bus subject cannot carry (a space, `*`, `>`) written `_`. The event's body names the group exactly.
func GroupTakeInKey(group string) string {
return "g." + strings.Map(func(r rune) rune {
if r == '*' || r == '>' || r <= ' ' || r == 0x7f {
return '_'
}
return r
}, group)
}
// TakeInEvent is the name of a take-in event of the given kind and key: `take-in.<pending|taken>.<key>`.
func TakeInEvent(kind, key string) (string, error) {
if kind != TakeInPending && kind != TakeInTaken {
return "", fmt.Errorf("delivery: the take-in kind %q is not pending or taken", kind)
}
if err := checkTakeInKey(key); err != nil {
return "", err
}
return TakeInStream + "." + kind + "." + key, nil
}
// ParseTakeInEvent is the inverse of TakeInEvent.
func ParseTakeInEvent(event string) (kind, key string, err error) {
parts := strings.SplitN(event, ".", 3)
if len(parts) != 3 || parts[0] != TakeInStream {
return "", "", fmt.Errorf("delivery: %q is not a take-in event, take-in.<pending|taken>.<key>", event)
}
if _, err := TakeInEvent(parts[1], parts[2]); err != nil {
return "", "", fmt.Errorf("delivery: %q is not a take-in event: %w", event, err)
}
return parts[1], parts[2], nil
}
func checkTakeInKey(key string) error {
if !strings.HasPrefix(key, "m.") && !strings.HasPrefix(key, "g.") {
return fmt.Errorf("delivery: the take-in key %q is neither m.<sequence> nor g.<group>", key)
}
for _, tok := range strings.Split(key, ".") {
if tok == "" || strings.ContainsAny(tok, "*> \t\r\n") {
return fmt.Errorf("delivery: the take-in key %q is not a bus subject's tokens", key)
}
}
return nil
}
// TakeIn is the body of a take-in event. Kind is the event's kind and Key its key. What says what waits: a revision
// (Key its oldest merge's PendingKey, WaitsFor the indexer build it names) or a group being gathered (Key its
// GroupTakeInKey, Members every approved member with its head, Merges the members held so far). Merges are the merges
// it holds, oldest first; Since is when the wait began. A `taken` event names the same once the wait ended; By and Why
// name a hand act (a hand take-in, or a drop, Dropped), and are carried by nothing else.
type TakeIn struct {
V int `json:"v"`
Kind string `json:"kind"`
Key string `json:"key"`
What string `json:"what"`
Merges []GroupMerge `json:"merges"`
Group string `json:"group,omitempty"`
Members []Head `json:"members,omitempty"`
WaitsFor string `json:"waits-for,omitempty"`
Since time.Time `json:"since"`
By string `json:"by,omitempty"`
Why string `json:"why,omitempty"`
Dropped bool `json:"dropped,omitempty"`
}
func (t TakeIn) check() error {
if _, err := TakeInEvent(t.Kind, t.Key); err != nil {
return err
}
if len(t.Merges) == 0 {
return fmt.Errorf("delivery: the take-in event %s names no merge", t.Key)
}
for _, m := range t.Merges {
if err := checkRepository(m.Repository); err != nil {
return err
}
if m.Merge == "" {
return fmt.Errorf("delivery: the take-in event %s names a merge of %s without its commit", t.Key, m.Repository)
}
}
if t.Since.IsZero() {
return fmt.Errorf("delivery: the take-in event %s does not say since when it waits", t.Key)
}
switch t.What {
case TakeInRevision:
if !strings.HasPrefix(t.Key, "m.") {
return fmt.Errorf("delivery: a revision's take-in key is its oldest merge's m.<sequence>, not %q", t.Key)
}
if t.Kind == TakeInPending && t.WaitsFor == "" {
return fmt.Errorf("delivery: a pending revision %s names the indexer build it waits for", t.Key)
}
case TakeInGroup:
if t.Group == "" {
return fmt.Errorf("delivery: a group's take-in event %s names its group", t.Key)
}
if t.Key != GroupTakeInKey(t.Group) {
return fmt.Errorf("delivery: the take-in key of the group %q is %s, not %s", t.Group, GroupTakeInKey(t.Group), t.Key)
}
if t.WaitsFor != "" {
return fmt.Errorf("delivery: a group being gathered waits for its end, not for an indexer build")
}
default:
return fmt.Errorf("delivery: a take-in event's what %q is not revision or group", t.What)
}
if t.Kind != TakeInTaken && (t.By != "" || t.Why != "" || t.Dropped) {
return fmt.Errorf("delivery: only a taken event names the hand act that ended a wait")
}
if (t.Why != "" || t.Dropped) && t.By == "" {
return fmt.Errorf("delivery: a hand act's take-in event names who acted (by)")
}
return nil
}
// EncodeTakeIn is t as JSON at Version; it refuses what its kind and what do not carry.
func EncodeTakeIn(t TakeIn) ([]byte, error) {
if err := t.check(); err != nil {
return nil, err
}
t.V = Version
return json.Marshal(t)
}
// DecodeTakeIn reads a take-in event's body; it refuses an unknown version (ErrUnknownVersion) and what EncodeTakeIn
// refuses.
func DecodeTakeIn(data []byte) (TakeIn, error) {
var t TakeIn
if err := decode(data, &t); err != nil {
return TakeIn{}, err
}
if err := t.check(); err != nil {
return TakeIn{}, err
}
return t, nil
}
// SameTakeIn says whether two take-in bodies are the same value.
func SameTakeIn(a, b TakeIn) bool { return reflect.DeepEqual(a, b) }