From f0c7d468c379114d7204af05aa214582934df70a Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 22:11:59 +0100 Subject: [PATCH] GATE A-33, A-34: a take-in.taken opens one delivery record and moves its branch's source, saying source-moved MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §151, §291, §592, A-33, A-34 (G-3, G-5, G-16, G-20) --- .../cmd/mesh-delivery-gate/consumer.go | 16 ++- .../cmd/mesh-delivery-gate/main.go | 2 +- .../cmd/mesh-delivery-gate/merges.go | 38 ++++--- .../cmd/mesh-delivery-gate/takenin.go | 98 +++++++++++++++++++ .../cmd/mesh-delivery-gate/takenin_test.go | 94 ++++++++++++++++++ 5 files changed, 224 insertions(+), 24 deletions(-) create mode 100644 modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin.go create mode 100644 modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin_test.go diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/consumer.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/consumer.go index ae74e5266..2559e6c2c 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/consumer.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/consumer.go @@ -120,6 +120,8 @@ type StepConsumer struct { Families []stepFamily Now func() time.Time Logf func(format string, a ...any) + // Taken opens the record of a take-in.taken and moves its branch's source (takenin.go), before its outcomes are applied. + Taken func(body json.RawMessage) error mu sync.Mutex // heard counts, per family, the events applied, refused and the last applied, since this owner started. @@ -180,6 +182,11 @@ func (c *StepConsumer) Hear(e stdio.Envelope) error { c.logf("[mesh-delivery] the record does not take %s: %s", e.Key, why) return nil } + if f.Event == takein.TakeInTakenEvent && c.Taken != nil { + if err := c.Taken(e.Body); err != nil { + return fmt.Errorf("the record of %s is not opened yet: %w", e.Key, err) + } + } id := e.Headers["x-event-id"] for _, o := range outcomes { if f.NodeScoped && o.Node == "" { @@ -267,10 +274,13 @@ func holds(r *Record, o StepOutcome) bool { // mergedAs are the commits the record's members merged as. func (r *Record) mergedAs() []string { - if r.Group == nil { - return nil - } var out []string + for _, c := range r.Heads { + out = append(out, c) + } + if r.Group == nil { + return out + } for _, p := range r.Group.Pulls { if p.Commit != "" { out = append(out, p.Commit) diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go index e3e8728d7..195bb851e 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -64,7 +64,7 @@ func main() { h.SaySeat = stdio.SeatPublish h.Records = &Records{Ledger: sdkLedger{}, Now: h.Now, Retention: h.recordRetention, Say: stdio.SeatPublish, Logf: logf} h.Old = kvOld{} - h.Consumer = &StepConsumer{Records: h.Records, Families: stepFamilies, Now: h.Now, Logf: logf} + h.Consumer = &StepConsumer{Records: h.Records, Families: stepFamilies, Now: h.Now, Logf: logf, Taken: h.TakenIn} l := &listening{now: "not yet: starting"} go run(h, l) if err := announceKept(stdio.Keeps); err != nil { diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go index 24ad47157..cf42260d3 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go @@ -49,9 +49,9 @@ func (kvOld) Get(state, key string) ([]byte, uint64, error) { // SourceStore is the gate's bucket source, one key per watched branch (design 47 G §15), moved by compare-and-set. type SourceStore interface { - // Move reads the source key of the gate's merge branch (MergeBranch) and writes what update makes of it, unless update - // says nothing is to be written; a write another writer won is read again and tried again. - Move(update func(prev *Source) (Source, bool)) error + // Set reads a branch's source key and writes what update makes of it, unless update says nothing is to be written, and answers + // the key's revision; a write another writer won is read again and tried again. + Set(branch string, update func(prev *Source) (Source, bool)) (uint64, error) // Read is a branch's source key as held and its revision on the bus; nil and 0 before it is first written. Read(branch string) (*Source, uint64, error) // Branches are the branch names of the source keys the gate keeps (§614, §690). @@ -80,42 +80,44 @@ func (kvMerges) Get(group string) (*Merge, error) { // kvSource is source over the runtime. type kvSource struct{} -func (kvSource) Move(update func(prev *Source) (Source, bool)) error { +func (kvSource) Set(branch string, update func(prev *Source) (Source, bool)) (uint64, error) { state := stdio.State("source") + key := mdc.SourceKey(keys.EscapeName(branch)) for range changeAttempts { - e, err := state.Get(mdc.SourceKey(MergeBranch)) + e, err := state.Get(key) if err != nil { - return err + return 0, err } var prev *Source + var rev uint64 if e != nil { p, err := mdc.DecodeSource(e.Value) if err != nil { - return fmt.Errorf("source.main is not readable: %w", err) + return 0, fmt.Errorf("source.%s is not readable: %w", branch, err) } - prev = &p + prev, rev = &p, e.Revision } next, write := update(prev) if !write { - return nil + return rev, nil } raw, err := mdc.EncodeSource(next) if err != nil { - return err + return 0, err } if e == nil { - _, err = state.Create(mdc.SourceKey(MergeBranch), json.RawMessage(raw)) + rev, err = state.Create(key, json.RawMessage(raw)) } else { - _, err = state.Update(mdc.SourceKey(MergeBranch), json.RawMessage(raw), e.Revision) + rev, err = state.Update(key, json.RawMessage(raw), e.Revision) } if err == nil { - return nil + return rev, nil } if !errors.Is(err, stdio.ErrChanged) { - return err + return 0, err } } - return fmt.Errorf("source.main was written by another writer first %d times: %w", changeAttempts, stdio.ErrChanged) + return 0, fmt.Errorf("source.%s was written by another writer first %d times: %w", branch, changeAttempts, stdio.ErrChanged) } func (kvSource) Read(branch string) (*Source, uint64, error) { @@ -189,11 +191,7 @@ func (h *Holder) SourceMain() map[string]any { if h.Source == nil { return map[string]any{"error": "no source bucket"} } - var held *Source - err := h.Source.Move(func(prev *Source) (Source, bool) { - held = prev - return Source{}, false - }) + held, _, err := h.Source.Read(MergeBranch) switch { case err != nil: return map[string]any{"error": err.Error()} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin.go new file mode 100644 index 000000000..e72e45a44 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin.go @@ -0,0 +1,98 @@ +package main + +import ( + "encoding/json" + "fmt" + "sort" + "strconv" + + mdc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" + meshcatalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// What a take-in.taken does at the gate (design 47 G A-33, A-34; §151, §291, §592): it opens one delivery record, keyed by the +// taken body's branch and source revision, and moves that branch's source revision, saying source-moved. Nothing else opens a +// record or moves a source key. + +// recordMilestone is the key of a taken revision's record: `.`, each escaped as a name (A-33). +func recordMilestone(branch string, source int64) string { + return keys.EscapeName(branch) + "." + keys.EscapeName(strconv.FormatInt(source, 10)) +} + +// TakenIn takes one take-in.taken body: the record of its source revision is opened (safe to repeat: a record there already +// is left as it is), its members are the entries' pull requests, its heads the merge commits the take-in names; then the +// branch's source key is written and source-moved said. A hand drop took nothing in, so it opens nothing. +func (h *Holder) TakenIn(body json.RawMessage) error { + var t meshcatalog.TakeInTaken + if err := json.Unmarshal(body, &t); err != nil { + return fmt.Errorf("a take-in.taken is not readable: %w", err) + } + if t.Dropped || len(t.Merges) == 0 || t.Branch == "" { + return nil + } + if h.Records != nil { + heads := map[string]string{} + for _, m := range t.Merges { + heads[mdc.ApprovedKey(m.Repository, int(m.Number))] = m.Merge + } + milestone := recordMilestone(t.Branch, t.Source) + r, _, err := h.Records.Open(milestone, heads, false) + if err != nil { + return fmt.Errorf("the record of %s cannot be opened: %w", milestone, err) + } + g := RecordGroup{Group: t.Group} + seen := map[string]bool{} + for _, e := range t.Entries { + if e.Repository == "" || seen[mdc.ApprovedKey(e.Repository, int(e.Pull))] { + continue + } + seen[mdc.ApprovedKey(e.Repository, int(e.Pull))] = true + g.Members = append(g.Members, mdc.ApprovedKey(e.Repository, int(e.Pull))) + } + sort.Strings(g.Members) + for _, m := range t.Merges { + g.Pulls = append(g.Pulls, RecordPull{Repository: m.Repository, Number: int(m.Number), Head: heads[mdc.ApprovedKey(m.Repository, int(m.Number))], + State: string(mdc.MergeMerged), Commit: m.Merge}) + } + if err := h.Records.SetGroup(r, g); err != nil { + return err + } + } + return h.moveSource(t) +} + +// moveSource writes the taken source revision of a branch: the merge commit of each repository it took in, the group that +// moved it, and says source-moved with the key's revision. +func (h *Holder) moveSource(t meshcatalog.TakeInTaken) error { + if h.Source == nil { + return nil + } + heads := map[string]string{} + for _, m := range t.Merges { + heads[m.Repository] = m.Merge + } + rev, err := h.Source.Set(t.Branch, func(prev *Source) (Source, bool) { + next := Source{Group: t.Group, Heads: heads} + if prev != nil { + next.Heads = map[string]string{} + for r, c := range prev.Heads { + next.Heads[r] = c + } + for r, c := range heads { + next.Heads[r] = c + } + } + return next, true + }) + if err != nil { + return fmt.Errorf("source.%s cannot be moved: %w", t.Branch, err) + } + if h.SaySeat == nil { + return nil + } + at := h.Now().UTC() + return h.SaySeat(seat.SubjectSourceMoved(keys.EscapeName(t.Branch)), + seat.SourceMovedEvent{Branch: t.Branch, Group: t.Group, Revision: int64(rev), At: &at}, "source-moved."+keys.EscapeName(t.Branch)+"."+strconv.FormatUint(rev, 10)) +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin_test.go new file mode 100644 index 000000000..307d546a0 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin_test.go @@ -0,0 +1,94 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// memSource is the gate's source bucket in memory: one value and revision per branch key. +type memSource struct { + vals map[string]Source + rev uint64 +} + +func (m *memSource) Set(branch string, update func(prev *Source) (Source, bool)) (uint64, error) { + var prev *Source + if p, ok := m.vals[branch]; ok { + prev = &p + } + next, write := update(prev) + if !write { + return m.rev, nil + } + if m.vals == nil { + m.vals = map[string]Source{} + } + m.vals[branch] = next + m.rev++ + return m.rev, nil +} +func (m *memSource) Read(branch string) (*Source, uint64, error) { + v, ok := m.vals[branch] + if !ok { + return nil, 0, nil + } + return &v, m.rev, nil +} +func (m *memSource) Branches() ([]string, error) { return nil, nil } + +// Design 47 G A-33, A-34: a take-in.taken opens one record keyed `.`, whose members are the entries' pull +// requests and whose heads are the merge commits it names, moves that branch's source key and says source-moved; heard twice +// it opens no second record; a hand drop opens nothing and moves nothing. +func TestATakeInTakenOpensOneRecordAndMovesItsBranchsSource(t *testing.T) { + now := time.Date(2026, 11, 11, 20, 0, 0, 0, time.UTC) + type said struct{ subject, id string } + var events []said + src := &memSource{} + h := &Holder{Now: func() time.Time { return now }, Source: src, + SaySeat: func(subject string, body any, id string) error { + events = append(events, said{subject, id}) + return nil + }} + h.Records = &Records{Ledger: newLedgerFake(t), Now: h.Now} + merge := strings.Repeat("a", 40) + body := json.RawMessage(`{"v":1,"kind":"taken","key":"g.feat","what":"group","group":"feat","branch":"main","source":7,` + + `"merges":[{"repository":"novox/app","number":3,"merge":"` + merge + `"}],` + + `"entries":[{"module":"app","commit":"` + merge + `","fingerprint":"f","repository":"novox/app","pull":3}],"since":"2026-11-11T19:00:00Z"}`) + c := &StepConsumer{Records: h.Records, Families: stepFamilies, Now: h.Now, Taken: h.TakenIn} + for i := 0; i < 2; i++ { + if err := c.Hear(stdio.Envelope{Key: "mesh-delivery-indexer.take-in.taken", Body: body}); err != nil { + t.Fatal(err) + } + } + kept, err := h.Records.Kept() + if err != nil || len(kept) != 1 { + t.Fatalf("records %v %v: one per take-in", kept, err) + } + r := kept[0] + if r.Milestone != keys.EscapeName("main")+".7" || r.Heads["novox/app@3"] != merge || r.Group == nil || + len(r.Group.Members) != 1 || r.Group.Members[0] != "novox/app@3" || r.Group.Group != "feat" { + t.Fatalf("the record %+v group %+v", r, r.Group) + } + if got := r.Outcomes[OutcomeKey("", "", "", "take-in")]; got.Outcome != OutcomeTakenIn { + t.Fatalf("the take-in is not the record's outcome: %+v", r.Outcomes) + } + if s := src.vals["main"]; s.Group != "feat" || s.Heads["novox/app"] != merge { + t.Fatalf("source.main %+v", s) + } + if len(events) != 2 || !strings.HasSuffix(events[0].subject, ".event.source-moved.main") { + t.Fatalf("said %+v", events) + } + drop := json.RawMessage(`{"v":1,"kind":"taken","key":"m.00000000000000000002","what":"revision","dropped":true,"branch":"main","source":8,` + + `"merges":[{"repository":"novox/lib","number":4,"merge":"` + merge + `"}],"since":"2026-11-11T19:00:00Z"}`) + if err := c.Hear(stdio.Envelope{Key: "mesh-delivery-indexer.take-in.taken", Body: drop}); err != nil { + t.Fatal(err) + } + if kept, _ := h.Records.Kept(); len(kept) != 1 || len(events) != 2 { + t.Fatalf("a hand drop opened %d records and said %d events", len(kept), len(events)) + } +}