diff --git a/go/delivery/source_test.go b/go/delivery/source_test.go index caf9809..e740eab 100644 --- a/go/delivery/source_test.go +++ b/go/delivery/source_test.go @@ -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..`, 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"}) diff --git a/go/delivery/takein.go b/go/delivery/takein.go new file mode 100644 index 0000000..4971796 --- /dev/null +++ b/go/delivery/takein.go @@ -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..`. 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.`, 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.`, 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..`. +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..", 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. nor g.", 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., 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) }