From 3f25e5b4da2818e37d9a0e9871e67fb4b688142c Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 14:35:57 +0100 Subject: [PATCH] The delivery record hears the step events through the gate's own consumer (milestone 47/G, W11.2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Design 47 (The delivery record): the outcomes arrive as events through the module's durable consumer, each applied once and only from its owner's account. A family per step event names the pattern the manifest consumes, its owner and whether its subject names a node; an event is applied only when its key names its owner and, for a node's event, its x-node is the node its subject names. It is taken once its outcome is in every open record holding it (by a member's merge commit, or by the entry of the record's graph difference), so one the ledger did not take is offered again; an unreadable one is said and taken. The indexer's take-in.taken is the first family; each other step's event joins the table with its owner's contract. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ (Joined from 4c4d1239.) Builds: design 47 G §34, §141 to §146 (GATE2) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-gate/consumer.go | 277 ++++++++++++++++++ .../cmd/mesh-delivery-gate/holder.go | 2 + .../cmd/mesh-delivery-gate/main.go | 9 +- .../cmd/mesh-delivery-gate/record.go | 106 ++++++- .../cmd/mesh-delivery-gate/record_test.go | 100 +++++++ modules/mesh-delivery-gate/go.mod | 1 + modules/mesh-delivery-gate/module.json | 16 +- 7 files changed, 505 insertions(+), 6 deletions(-) create mode 100644 modules/mesh-delivery-gate/cmd/mesh-delivery-gate/consumer.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 new file mode 100644 index 000000000..6c70dbc61 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/consumer.go @@ -0,0 +1,277 @@ +package main + +// The delivery record's consumer (design 47 *The delivery record*: "The outcomes arrive as events, through a durable +// consumer of the gate's own"). The gate's manifest `consumes` each step event of *What each step says* and the waits' +// `wait-held` and `wait-ended`; the runtime binds the module's durable consumer for them (stdio.Subscribe), which resumes +// from the last event the gate took after a restart, and offers an event again until the gate took it. The gate takes an +// event only once its outcome is in every record that holds it, so an event the ledger could not take is offered again; +// one applied twice changes nothing (Records.Outcome); and one is applied only when it comes from its owner's account: +// the event's key is its owner and its event recovered from the subject, which the bus grants its owner's account alone +// (ADR 0363 §2), and an event a node publishes names that node in its subject and in its x-node header. + +import ( + "encoding/json" + "fmt" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// StepOutcome is what one step event says of one entry, or of the record's merge when it names no module: the merge +// commits it is for (an event of the whole change), or the entry's module, fingerprint and node. +type StepOutcome struct { + Commits []string + Module string + Fingerprint string + Node string + Step string + Outcome string + Why string + At time.Time +} + +// stepFamily is one family of events the record is built from: the pattern its manifest consumes it by, its owner +// (the seat or module whose account alone publishes it), whether its subject names the node after the event, and how +// its body reads as outcomes. +type stepFamily struct { + Pattern string + Owner string + // Event is the event's name after its owner; the node is the token after it when NodeScoped. + Event string + NodeScoped bool + Decode func(body json.RawMessage, at time.Time) ([]StepOutcome, error) +} + +// The step events on main the record is built from. A family is added here, with its `consumes` line, when its owner +// publishes it (design 47 *What each step says*); each other step's event is its owner's to add in the same milestone. +var stepFamilies = []stepFamily{ + // An entry taken in (the indexer's own `take-in.taken`, ADR 0633 §1): the merges a source revision took in. + {Pattern: indexerModule + "." + takein.TakeInTakenEvent, Owner: indexerModule, Event: takein.TakeInTakenEvent, Decode: decodeTakenIn}, +} + +// indexerModule is the holder of the seat mesh-catalog whose own events the record hears. +const indexerModule = "mesh-delivery-indexer" + +// stepPatterns are the patterns of every family, in the order the manifest consumes them. +func stepPatterns() []string { + out := make([]string, 0, len(stepFamilies)) + for _, f := range stepFamilies { + out = append(out, f.Pattern) + } + return out +} + +// decodeTakenIn reads a `take-in.taken`: each merge taken in is an outcome of the record that merged it, unless the wait +// was dropped by hand, which took nothing in. +func decodeTakenIn(body json.RawMessage, at time.Time) ([]StepOutcome, error) { + t, err := takein.DecodeTakeIn(body) + if err != nil { + return nil, err + } + if t.Kind != takein.TakeInTaken { + return nil, fmt.Errorf("a take-in.taken of kind %q", t.Kind) + } + outcome, why := OutcomeTakenIn, t.Why + if t.Dropped { + outcome, why = OutcomeHeld, "its take-in was dropped: "+t.Why + } + var commits []string + for _, m := range t.Merges { + commits = append(commits, m.Merge) + } + if len(commits) == 0 { + return nil, nil + } + return []StepOutcome{{Commits: commits, Step: "take-in", Outcome: outcome, Why: why, At: at}}, nil +} + +// StepConsumer applies the step events to the records. +type StepConsumer struct { + Records *Records + Families []stepFamily + Now func() time.Time + Logf func(format string, a ...any) + + mu sync.Mutex + // heard counts, per family, the events applied, refused and the last applied, since this owner started. + heard map[string]*familyHeard +} + +type familyHeard struct { + Applied int `json:"applied"` + Refused int `json:"refused"` + Last time.Time `json:"last,omitzero"` + Why string `json:"last-refused,omitempty"` +} + +func (c *StepConsumer) now() time.Time { + if c.Now == nil { + return time.Now().UTC() + } + return c.Now().UTC() +} + +func (c *StepConsumer) logf(format string, a ...any) { + if c.Logf != nil { + c.Logf(format, a...) + } +} + +// familyOf is the family an event's key belongs to. +func (c *StepConsumer) familyOf(key string) (stepFamily, bool) { + for _, f := range c.Families { + if stdio.TopicMatches(f.Pattern, key) || (f.NodeScoped && strings.HasPrefix(key, f.Owner+"."+f.Event+".")) { + return f, true + } + } + return stepFamily{}, false +} + +// Takes is whether the consumer takes events of this key. +func (c *StepConsumer) Takes(key string) bool { _, ok := c.familyOf(key); return ok } + +// Hear applies one event. It answers an error only when an outcome could not be written, so the event is offered +// again; an event of no family, from another account than its owner's, or that cannot be read is said and taken, since +// offered again it would be refused again. +func (c *StepConsumer) Hear(e stdio.Envelope) error { + f, ok := c.familyOf(e.Key) + if !ok { + return nil + } + if why := ownerRefusal(f, e); why != "" { + c.count(f, false, why) + c.logf("[mesh-delivery] the record does not take %s: %s", e.Key, why) + return nil + } + at := c.now() + outcomes, err := f.Decode(e.Body, at) + if err != nil { + why := fmt.Sprintf("its body is not readable: %v", err) + c.count(f, false, why) + c.logf("[mesh-delivery] the record does not take %s: %s", e.Key, why) + return nil + } + id := e.Headers["x-event-id"] + for _, o := range outcomes { + if f.NodeScoped && o.Node == "" { + o.Node = e.Node + } + if err := c.apply(o, id); err != nil { + return fmt.Errorf("the outcome of %s is not recorded yet: %w", e.Key, err) + } + } + c.count(f, true, "") + return nil +} + +// ownerRefusal is why an event did not come from its owner's account, or "": its key names its owner, and an event of a +// node names the node its x-node header says. +func ownerRefusal(f stepFamily, e stdio.Envelope) string { + rest, ok := strings.CutPrefix(e.Key, f.Owner+".") + if !ok { + return fmt.Sprintf("it is not %s's own event", f.Owner) + } + if !f.NodeScoped { + return "" + } + after, ok := strings.CutPrefix(rest, f.Event+".") + node, _, _ := strings.Cut(after, ".") + if !ok || node == "" { + return "its subject names no node" + } + if e.Node != node { + return fmt.Sprintf("its subject names the node %s and it came from %q", node, e.Node) + } + return "" +} + +// apply writes an outcome into every open record that holds it: by its merge commits, or by its entry. +func (c *StepConsumer) apply(o StepOutcome, event string) error { + records, err := c.Records.Current() + if err != nil { + return err + } + for _, r := range records { + if r.Ended != nil || !holds(r, o) { + continue + } + ro := RecordOutcome{Module: o.Module, Fingerprint: o.Fingerprint, Node: o.Node, Step: o.Step, Outcome: o.Outcome, + Why: o.Why, At: o.At, Event: event} + if _, err := c.Records.Outcome(r, ro); err != nil { + return err + } + } + return nil +} + +// holds is whether a record is the one an outcome is for: one of its members merged as a commit the outcome names, or +// its difference holds the entry. +func holds(r *Record, o StepOutcome) bool { + if len(o.Commits) > 0 { + for _, m := range r.mergedAs() { + for _, c := range o.Commits { + if sameCommit(m, c) { + return true + } + } + } + return false + } + _, ok := r.Entries[EntryKey(o.Module, o.Fingerprint, o.Node)] + if !ok && o.Node != "" { + _, ok = r.Entries[EntryKey(o.Module, o.Fingerprint, "")] + } + return ok +} + +// 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 _, p := range r.Group.Pulls { + if p.Commit != "" { + out = append(out, p.Commit) + } + } + return out +} + +func (c *StepConsumer) count(f stepFamily, applied bool, why string) { + c.mu.Lock() + defer c.mu.Unlock() + if c.heard == nil { + c.heard = map[string]*familyHeard{} + } + h := c.heard[f.Pattern] + if h == nil { + h = &familyHeard{} + c.heard[f.Pattern] = h + } + if applied { + h.Applied++ + h.Last = c.now() + } else { + h.Refused++ + h.Why = why + } +} + +// Heard is, per family, what the consumer took since this owner started. +func (c *StepConsumer) Heard() map[string]familyHeard { + c.mu.Lock() + defer c.mu.Unlock() + out := map[string]familyHeard{} + for _, f := range c.Families { + if h := c.heard[f.Pattern]; h != nil { + out[f.Pattern] = *h + } else { + out[f.Pattern] = familyHeard{} + } + } + return out +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go index 5c91bee57..470b68fde 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go @@ -40,6 +40,8 @@ type Holder struct { Owed OwedStore // Records keeps the delivery records on the gate's ledger (record.go, ADR 0656 §7); nil keeps none. Records *Records + // Consumer applies the step events to the records (consumer.go); nil applies none. + Consumer *StepConsumer // Pulls keeps the pull requests' keys on the bus (pullkey.go); nil keeps none. Pulls PullKeys // Say says a condition to the operator, and Conditions keeps which the gate raised (conditions.go); nil says none. 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 2d150687a..939097202 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -32,9 +32,10 @@ func (l *listening) set(s string) { l.mu.Lock(); l.now = s; l.mu.Unlock() } func (l *listening) get() string { l.mu.Lock(); defer l.mu.Unlock(); return l.now } // The events this owner follows: the forge's pull requests. The forge's status events say that a check answered (events.go). -var followed = []string{"gitea.pull.updated", "gitea.pull.merged", "gitea.pull.closed", +// The step events the delivery record is built from follow (consumer.go). +var followed = append([]string{"gitea.pull.updated", "gitea.pull.merged", "gitea.pull.closed", "gitea.forge.status", "gitea.forge.pull-request.edited", "gitea.forge.pull-request.closed", - forgePattern, ConfigChangedEvent, writableEvent} + forgePattern, ConfigChangedEvent, writableEvent}, stepPatterns()...) // forgePattern is the forge bridge's messages on FORGE (novox/hq ADR 0437 §8): the gate follows a merge and the refusal of one. const forgePattern = "mesh-delivery-forge-bridge.forge.**" @@ -74,6 +75,7 @@ func main() { } h.init() h.Records = &Records{Ledger: sdkLedger{}, Now: h.Now, Retention: h.recordRetention} + h.Consumer = &StepConsumer{Records: h.Records, Families: stepFamilies, Now: h.Now, Logf: logf} l := &listening{now: "not yet: starting"} go run(h, l) if err := announceKept(stdio.Keeps); err != nil { @@ -186,6 +188,9 @@ func handle(h *Holder, e stdio.Envelope) error { } func take(h *Holder, e stdio.Envelope) error { + if h.Consumer != nil && h.Consumer.Takes(e.Key) { + return h.Consumer.Hear(e) + } if strings.HasPrefix(e.Key, "mesh-delivery-forge-bridge.forge.") { h.Forged(e.Key, e.Body) return nil diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go index b838552d2..aae76a812 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go @@ -18,6 +18,7 @@ import ( "regexp" "sort" "strings" + "sync" "time" stdio "git.novox.be/novox/mesh-sdk/go" @@ -45,6 +46,7 @@ const ( OpOpen = "open" // set-fields: the milestone, the heads, when, forge-made, the form OpGroup = "group" // set-fields: the group's state OpTrace = "trace" // set-fields: the trace + OpEntry = "entry" // set-entry into entries: one entry of the graph difference, by its module, fingerprint and node OpOutcome = "outcome" // set-entry into outcomes, by the entry, the node and the step OpAway = "away" // set-entry into away, by node OpBack = "back" // remove from away, by node @@ -57,7 +59,7 @@ const ( ) // RecordOperations are a record's operations, in the order the manifest's type allows them. -var RecordOperations = []string{OpOpen, OpGroup, OpTrace, OpOutcome, OpAway, OpBack, OpPin, OpFree, OpEnd, OpRemove} +var RecordOperations = []string{OpOpen, OpGroup, OpTrace, OpEntry, OpOutcome, OpAway, OpBack, OpPin, OpFree, OpEnd, OpRemove} // HeldOperations are the month's list's operations. var HeldOperations = []string{OpHold, OpDrop} @@ -84,6 +86,8 @@ type Record struct { Group *RecordGroup `json:"group,omitempty"` Trace *RecordTrace `json:"trace,omitempty"` + // Entries are the entries of the graph difference the record delivers, by EntryKey. + Entries map[string]RecordEntry `json:"entries,omitempty"` // Outcomes are what happened to each entry on each node, by OutcomeKey. Outcomes map[string]RecordOutcome `json:"outcomes,omitempty"` // Away are the nodes away, by node, each with since when. @@ -152,6 +156,20 @@ type RecordOutcome struct { Event string `json:"event,omitempty"` } +// RecordEntry is one entry of the graph difference: a module at a fingerprint, on a node (none for the catalog half's +// entry, which no node runs). +type RecordEntry struct { + Key string `json:"key"` + Module string `json:"module"` + Fingerprint string `json:"fingerprint"` + Node string `json:"node,omitempty"` +} + +// EntryKey is the key of an entry: its module, its fingerprint and its node. +func EntryKey(module, fingerprint, node string) string { + return strings.Join([]string{module, fingerprint, node}, "|") +} + // OutcomeKey is the key of an entry's outcome: its module, its fingerprint, its node and its step, each a token. func OutcomeKey(module, fingerprint, node, step string) string { return strings.Join([]string{module, fingerprint, node, step}, "|") @@ -305,6 +323,67 @@ type Records struct { Retention func() time.Duration // Freed is told each revision a removed record pinned, to say as `freed` (design 47); nil says none. Freed func(r *Record, p PinnedRevision) + + // held are the records kept, by instance, read once from the ledger (Current) and kept as the gate writes them: the + // gate is their one writer. + mu sync.Mutex + held map[string]*Record + loaded bool +} + +// Current is every record kept and not ended, oldest first: read from the ledger at the first call, then as the gate +// writes them. +func (rs *Records) Current() ([]*Record, error) { + if err := rs.load(); err != nil { + return nil, err + } + rs.mu.Lock() + defer rs.mu.Unlock() + var out []*Record + for _, r := range rs.held { + if r.Ended == nil { + out = append(out, r) + } + } + sort.SliceStable(out, func(i, j int) bool { return out[i].Opened.Before(out[j].Opened) }) + return out, nil +} + +func (rs *Records) load() error { + rs.mu.Lock() + loaded := rs.loaded + rs.mu.Unlock() + if loaded { + return nil + } + list, err := rs.List() + if err != nil { + return err + } + rs.mu.Lock() + defer rs.mu.Unlock() + if !rs.loaded { + rs.held = map[string]*Record{} + for _, r := range list { + rs.held[r.Instance] = r + } + rs.loaded = true + } + return nil +} + +func (rs *Records) keep(r *Record) { + rs.mu.Lock() + defer rs.mu.Unlock() + if rs.loaded { + rs.held[r.Instance] = r + } +} + +func (rs *Records) forget(instance string) { + rs.mu.Lock() + defer rs.mu.Unlock() + delete(rs.held, instance) } func (rs *Records) now() time.Time { @@ -368,6 +447,9 @@ func (rs *Records) Open(milestone string, heads map[string]string, forgeMade boo return nil, false, fmt.Errorf("the record %s is opened but not listed: %w", inst, err) } r, err = rs.Read(inst) + if err == nil && r != nil { + rs.keep(r) + } return r, err == nil && r != nil, err } @@ -443,6 +525,7 @@ func (rs *Records) Sweep(now time.Time) error { if _, err := rs.Ledger.Apply(e.Instance, OpRemove, map[string]any{}); err != nil { return err } + rs.forget(e.Instance) if rs.Freed != nil { for _, p := range sortedPins(r) { rs.Freed(r, p) @@ -482,8 +565,8 @@ func (rs *Records) SetGroup(r *Record, g RecordGroup) error { // Outcome writes one entry's outcome on one node, unless the record holds it already: an event applied twice changes // nothing (design 47). Applied says it was written now. func (rs *Records) Outcome(r *Record, o RecordOutcome) (applied bool, err error) { - if o.Module == "" || o.Step == "" || o.Outcome == "" { - return false, fmt.Errorf("an outcome names its module, its step and what happened (%+v)", o) + if o.Step == "" || o.Outcome == "" { + return false, fmt.Errorf("an outcome names its step and what happened (%+v)", o) } o.Key = OutcomeKey(o.Module, o.Fingerprint, o.Node, o.Step) if had, ok := r.Outcomes[o.Key]; ok && (had.Event != "" && had.Event == o.Event || sameOutcome(had, o)) { @@ -500,6 +583,23 @@ func (rs *Records) Outcome(r *Record, o RecordOutcome) (applied bool, err error) return true, nil } +// AddEntry writes one entry of the graph difference into the record, unless it holds it. +func (rs *Records) AddEntry(r *Record, e RecordEntry) error { + e.Key = EntryKey(e.Module, e.Fingerprint, e.Node) + if had, ok := r.Entries[e.Key]; ok && had == e { + return nil + } + seq, err := rs.Ledger.Apply(r.Instance, OpEntry, e) + if err != nil { + return err + } + if r.Entries == nil { + r.Entries = map[string]RecordEntry{} + } + r.Entries[e.Key], r.Sequence = e, seq + return nil +} + // sameOutcome is whether two outcomes say the same, whenever and from whichever event they were taken. func sameOutcome(a, b RecordOutcome) bool { a.At, b.At, a.Event, b.Event = time.Time{}, time.Time{}, "", "" diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go index b101b362f..3d1ac5f5c 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go @@ -10,6 +10,7 @@ import ( "testing" "time" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/catalogue" ) @@ -303,3 +304,102 @@ func TestTheManifestsLedgerTypesAreTheRecordsOperations(t *testing.T) { t.Error("the records ledger is not capped at 768 MiB") } } + +// W11.2 (design 47 *The delivery record*): the record's outcomes arrive as events through the gate's consumer: each +// applied once, only from its owner's account (its key's owner, and for a node's event the node its subject names), an +// unreadable one taken and said, and one the ledger could not take offered again. +func TestTheRecordsOutcomesArriveThroughTheGatesConsumer(t *testing.T) { + at := time.Date(2026, 11, 9, 21, 0, 0, 0, time.UTC) + merge := strings.Repeat("a", 40) + setup := func(t *testing.T) (*ledgerFake, *Records, *Record, *StepConsumer) { + l := newLedgerFake(t) + rs := &Records{Ledger: l, Now: func() time.Time { return at }} + r, _, err := rs.Open("47/G", map[string]string{"novox/app@111111111111": strings.Repeat("1", 40)}, false) + if err != nil { + t.Fatal(err) + } + if err := rs.SetGroup(r, RecordGroup{Group: "feat/x", Pulls: []RecordPull{{Repository: "novox/app", Number: 7, + Head: strings.Repeat("1", 40), State: "merged", Commit: merge}}}); err != nil { + t.Fatal(err) + } + if err := rs.AddEntry(r, RecordEntry{Module: "mesh-x", Fingerprint: "src2:f1", Node: "ace"}); err != nil { + t.Fatal(err) + } + running := stepFamily{Pattern: "node-engine.running.*.*", Owner: "node-engine", Event: "running", NodeScoped: true, + Decode: func(body json.RawMessage, at time.Time) ([]StepOutcome, error) { + var b struct{ Module, Fingerprint string } + if err := json.Unmarshal(body, &b); err != nil { + return nil, err + } + return []StepOutcome{{Module: b.Module, Fingerprint: b.Fingerprint, Step: "run", Outcome: OutcomeRunning, At: at}}, nil + }} + c := &StepConsumer{Records: rs, Families: append(slices.Clone(stepFamilies), running), Now: func() time.Time { return at }} + return l, rs, r, c + } + taken := func(id string) stdio.Envelope { + body := `{"v":1,"kind":"taken","key":"m.00000000000000000001","what":"revision","merges":[{"repository":"novox/app","number":7,"merge":"` + + merge + `"}],"since":"2026-11-09T20:00:00Z"}` + return stdio.Envelope{Key: "mesh-delivery-indexer.take-in.taken", Body: json.RawMessage(body), Headers: map[string]string{"x-event-id": id}} + } + t.Run("a take-in of the record's merge is its outcome, once", func(t *testing.T) { + l, rs, r, c := setup(t) + if err := c.Hear(taken("e1")); err != nil { + t.Fatal(err) + } + n := len(l.history) + if err := c.Hear(taken("e1")); err != nil || len(l.history) != n { + t.Fatalf("the same event again: %v, %d operations (was %d)", err, len(l.history), n) + } + back, _ := rs.Read(r.Instance) + if got := back.Outcomes[OutcomeKey("", "", "", "take-in")]; got.Outcome != OutcomeTakenIn || got.Event != "e1" { + t.Fatalf("outcomes %+v", back.Outcomes) + } + if c.Heard()["mesh-delivery-indexer.take-in.taken"].Applied != 2 { + t.Fatalf("heard %+v", c.Heard()) + } + }) + t.Run("a node's event from its own node is the entry's outcome; from another node it is refused", func(t *testing.T) { + _, rs, r, c := setup(t) + e := stdio.Envelope{Key: "node-engine.running.ace.mesh-x", Node: "g14", Body: json.RawMessage(`{"module":"mesh-x","fingerprint":"src2:f1"}`)} + if err := c.Hear(e); err != nil { + t.Fatal(err) + } + if back, _ := rs.Read(r.Instance); len(back.Outcomes) != 0 { + t.Fatalf("an event naming ace from g14 is applied: %+v", back.Outcomes) + } + if h := c.Heard()["node-engine.running.*.*"]; h.Refused != 1 || !strings.Contains(h.Why, "names the node ace") { + t.Fatalf("heard %+v", h) + } + e.Node = "ace" + if err := c.Hear(e); err != nil { + t.Fatal(err) + } + back, _ := rs.Read(r.Instance) + if got := back.Outcomes[OutcomeKey("mesh-x", "src2:f1", "ace", "run")]; got.Outcome != OutcomeRunning { + t.Fatalf("outcomes %+v", back.Outcomes) + } + }) + t.Run("an unreadable event is taken and said; one the ledger cannot take is offered again", func(t *testing.T) { + l, _, _, c := setup(t) + bad := taken("e2") + bad.Body = json.RawMessage(`{"v":1,"kind":"pending"}`) + if err := c.Hear(bad); err != nil { + t.Fatalf("an unreadable event is offered again: %v", err) + } + if h := c.Heard()["mesh-delivery-indexer.take-in.taken"]; h.Refused != 1 { + t.Fatalf("heard %+v", h) + } + l.failOn = OpOutcome + if err := c.Hear(taken("e3")); err == nil { + t.Fatal("an outcome the ledger did not take is taken") + } + }) + t.Run("the manifest consumes every family", func(t *testing.T) { + m, _ := readManifest(t) + for _, f := range stepFamilies { + if !slices.Contains(m.Consumes, f.Pattern) { + t.Errorf("the manifest does not consume %s", f.Pattern) + } + } + }) +} diff --git a/modules/mesh-delivery-gate/go.mod b/modules/mesh-delivery-gate/go.mod index 1ff0a9a76..11b37039e 100644 --- a/modules/mesh-delivery-gate/go.mod +++ b/modules/mesh-delivery-gate/go.mod @@ -9,6 +9,7 @@ require ( git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-gate/module.json b/modules/mesh-delivery-gate/module.json index 0c48589f9..d7aa47e5f 100644 --- a/modules/mesh-delivery-gate/module.json +++ b/modules/mesh-delivery-gate/module.json @@ -28,7 +28,8 @@ "gitea.forge.pull-request.closed", "mesh-delivery-forge-bridge.forge.**", "mesh-config.config-changed", - "gitea.repo.writable" + "gitea.repo.writable", + "mesh-delivery-indexer.take-in.taken" ], "emits": [ "transition", @@ -140,6 +141,19 @@ ], "sequence": true }, + { + "name": "entry", + "effect": "set-entry", + "fields": [ + "key", + "module", + "fingerprint", + "node" + ], + "into": "entries", + "key": "key", + "sequence": true + }, { "name": "outcome", "effect": "set-entry",