The delivery record hears the step events through the gate's own consumer (milestone 47/G, W11.2)
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
This commit is contained in:
@@ -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
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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{}, "", ""
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user