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 6c70dbc61..33f55c13d 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/consumer.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/consumer.go @@ -188,14 +188,18 @@ func ownerRefusal(f stepFamily, e stdio.Envelope) string { return "" } -// apply writes an outcome into every open record that holds it: by its merge commits, or by its entry. +// apply writes an outcome into every record that holds it, by its merge commits or by its entry: an open record, or one +// that ended without the node the outcome comes from (a node away that wakes adds its outcomes, and is no longer shown +// away). An open record whose present nodes then all run their entries ends. func (c *StepConsumer) apply(o StepOutcome, event string) error { - records, err := c.Records.Current() + records, err := c.Records.Kept() if err != nil { return err } for _, r := range records { - if r.Ended != nil || !holds(r, o) { + _, fromAway := r.Away[o.Node] + fromAway = fromAway && o.Node != "" + if (r.Ended != nil && !fromAway) || !holds(r, o) { continue } ro := RecordOutcome{Module: o.Module, Fingerprint: o.Fingerprint, Node: o.Node, Step: o.Step, Outcome: o.Outcome, @@ -203,6 +207,14 @@ func (c *StepConsumer) apply(o StepOutcome, event string) error { if _, err := c.Records.Outcome(r, ro); err != nil { return err } + if fromAway { + if err := c.Records.NodeBack(r, o.Node); err != nil { + return err + } + } + if _, err := c.Records.EndIfRun(r); err != nil { + return err + } } return nil } 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 64df2e448..b22b758d5 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go @@ -42,6 +42,8 @@ type Holder struct { Records *Records // Consumer applies the step events to the records (consumer.go); nil applies none. Consumer *StepConsumer + // SaySeat says an event of the seat mesh-delivery (recordevents.go): heads-replaced here; nil says none. + SaySeat SeatSay // 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. @@ -416,6 +418,7 @@ func (h *Holder) supersedeOlder(p PullEvent) (replaced string) { if old.Repository == p.repository() && old.Number == p.Number && !sameCommit(old.Commit, p.HeadSHA) && !old.State.Final() && old.MergedAs == "" { old.NewerHead = p.HeadSHA + h.sayHeadsReplaced(askGroup(old), h.candidateOf(old), old, p.HeadSHA, false) h.settleAndKeep(old) if old.Group != "" { h.leaveGroup(old) @@ -426,6 +429,17 @@ func (h *Holder) supersedeOlder(p PullEvent) (replaced string) { return replaced } +// candidateOf are the members whose heads make the candidate a delivery is judged in: its group's, or itself alone. +// Under the lock. +func (h *Holder) candidateOf(d *Delivery) []*Delivery { + if g := h.groups[d.Group]; d.Group != "" && g != nil { + if members := h.membersOf(g); len(members) > 0 { + return members + } + } + return []*Delivery{d} +} + // fillFromAnnouncement fills a delivery made from its check's verdict before its pull request's head was heard (issue // 1726: no base, branch, title or URLs) from an announcement or an edit of that pull request, and then // does what a first announcement does: the older heads of the pull request are superseded, and the delivery takes @@ -880,6 +894,7 @@ func (h *Holder) PullClosed(p PullEvent) { } if !d.State.Final() && d.MergedAs == "" { d.ClosedUnmerged = true + h.sayHeadsReplaced(askGroup(d), h.candidateOf(d), d, "", true) h.settleAndKeep(d) if d.Group != "" { h.leaveGroup(d) 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 068c09d19..74f682a7e 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -73,7 +73,8 @@ func main() { Conditions: kvConditions{}, } h.init() - h.Records = &Records{Ledger: sdkLedger{}, Now: h.Now, Retention: h.recordRetention} + h.SaySeat = stdio.SeatPublish + h.Records = &Records{Ledger: sdkLedger{}, Now: h.Now, Retention: h.recordRetention, Say: stdio.SeatPublish} h.Merges = recordedMerges{MergeStore: kvMerges{}, records: h.Records, members: h.groupMembers} h.Consumer = &StepConsumer{Records: h.Records, Families: stepFamilies, Now: h.Now, Logf: logf} l := &listening{now: "not yet: starting"} 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 aae76a812..0c0934532 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go @@ -321,8 +321,8 @@ type Records struct { Now func() time.Time // Retention is how long a record is kept; nil keeps DefaultRecordRetention. 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) + // Say says the seat mesh-delivery's events of the records: pinned, freed, record-ended (recordevents.go); nil says none. + Say SeatSay // 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. @@ -349,6 +349,21 @@ func (rs *Records) Current() ([]*Record, error) { return out, nil } +// Kept is every record kept, ended or not, oldest first. +func (rs *Records) Kept() ([]*Record, error) { + if err := rs.load(); err != nil { + return nil, err + } + rs.mu.Lock() + defer rs.mu.Unlock() + out := make([]*Record, 0, len(rs.held)) + for _, r := range rs.held { + 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 @@ -522,15 +537,13 @@ func (rs *Records) Sweep(now time.Time) error { return err } if r != nil { + if err := rs.sayFreedAll(r, "the record is past its retention"); err != nil { + return err + } 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) - } - } } if _, err := rs.Ledger.Apply(m, OpDrop, map[string]any{"instance": e.Instance}); err != nil { return err @@ -562,6 +575,19 @@ func (rs *Records) SetGroup(r *Record, g RecordGroup) error { return nil } +// SetTrace writes the record's trace, unless it holds that trace already. +func (rs *Records) SetTrace(r *Record, t RecordTrace) error { + if r.Trace != nil && reflect.DeepEqual(*r.Trace, t) { + return nil + } + seq, err := rs.Ledger.Apply(r.Instance, OpTrace, map[string]any{"trace": t}) + if err != nil { + return err + } + r.Trace, r.Sequence = &t, seq + return nil +} + // 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) { 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 c1b11d87a..9e141b073 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 @@ -5,12 +5,14 @@ import ( "fmt" "os" "path/filepath" + "reflect" "slices" "strings" "testing" "time" mdc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/catalogue" ) @@ -230,9 +232,15 @@ func TestTheDeliveryRecordIsALedgerInstanceKeyedByMilestoneAndHeads(t *testing.T now := at var freed []string rs := &Records{Ledger: l, Now: func() time.Time { return now }, Retention: func() time.Duration { return 90 * 24 * time.Hour }, - Freed: func(r *Record, p PinnedRevision) { freed = append(freed, r.Milestone+" "+p.Revision) }} + Say: func(subject string, body any, id string) error { + if subject == meshdelivery.SubjectFreed { + p := body.(meshdelivery.Pins) + freed = append(freed, fmt.Sprint(p.Milestone, " ", p.Catalog, " ", p.Why)) + } + return nil + }} old, _, _ := rs.Open("46/F", heads, false) - if _, err := l.Apply(old.Instance, OpPin, PinnedRevision{Revision: "cat-1", Half: "catalog"}); err != nil { + if err := rs.Pin(old, HalfCatalog, 1); err != nil { t.Fatal(err) } now = at.Add(89 * 24 * time.Hour) @@ -247,7 +255,7 @@ func TestTheDeliveryRecordIsALedgerInstanceKeyedByMilestoneAndHeads(t *testing.T if r, _ := rs.Read(old.Instance); r != nil { t.Fatal("a record past its retention is kept") } - if !slices.Equal(freed, []string{"46/F cat-1"}) { + if !slices.Equal(freed, []string{"46/F [1] the record is past its retention"}) { t.Fatalf("freed %v", freed) } list, _ := rs.List() @@ -256,6 +264,60 @@ func TestTheDeliveryRecordIsALedgerInstanceKeyedByMilestoneAndHeads(t *testing.T } }) + // W11.9: pinned and freed, said under an id of what they say; only a catalog or an assignment revision is pinned. + t.Run("a pin is written once and said as pinned; a free says freed with why", func(t *testing.T) { + l := newLedgerFake(t) + type said struct { + subject, id string + pins meshdelivery.Pins + } + var got []said + rs := &Records{Ledger: l, Now: func() time.Time { return at }, Say: func(subject string, body any, id string) error { + got = append(got, said{subject, id, body.(meshdelivery.Pins)}) + return nil + }} + r, _, _ := rs.Open("47/G", heads, false) + for range 2 { + if err := rs.Pin(r, HalfAssignment, 9); err != nil { + t.Fatal(err) + } + } + if err := rs.Pin(r, HalfCatalog, 9); err != nil { + t.Fatal(err) + } + pins := 0 + for _, op := range l.history { + if strings.HasSuffix(op, " "+OpPin) { + pins++ + } + } + back, _ := rs.Read(r.Instance) + if pins != 2 || len(back.Pinned) != 2 { + t.Fatalf("%d pins written, the record holds %+v", pins, back.Pinned) + } + if len(got) != 3 || !reflect.DeepEqual(got[0], got[1]) || got[0].subject != meshdelivery.SubjectPinned || + !slices.Equal(got[0].pins.Assignment, []int64{9}) || got[2].pins.Catalog[0] != 9 || got[0].pins.Record != r.Instance { + t.Fatalf("said %+v", got) + } + got = nil + if err := rs.Free(back, HalfAssignment, 9, "removed by hand"); err != nil { + t.Fatal(err) + } + if err := rs.Free(back, HalfAssignment, 9, "again"); err != nil { + t.Fatal(err) + } + if len(got) != 1 || got[0].subject != meshdelivery.SubjectFreed || got[0].pins.Why != "removed by hand" || + !slices.Equal(got[0].pins.Assignment, []int64{9}) { + t.Fatalf("said %+v", got) + } + if again, _ := rs.Read(r.Instance); len(again.Pinned) != 1 { + t.Fatalf("pinned after the free %+v", again.Pinned) + } + if err := rs.Pin(r, "trunk", 1); err == nil { + t.Fatal("a pin of no half is taken") + } + }) + t.Run("the setting record-retention, in days", func(t *testing.T) { dir := t.TempDir() for _, c := range []struct { @@ -332,7 +394,7 @@ func TestTheRecordsOutcomesArriveThroughTheGatesConsumer(t *testing.T) { 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 + return []StepOutcome{{Module: b.Module, Fingerprint: b.Fingerprint, Step: StepRunning, 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 @@ -376,10 +438,67 @@ func TestTheRecordsOutcomesArriveThroughTheGatesConsumer(t *testing.T) { t.Fatal(err) } back, _ := rs.Read(r.Instance) - if got := back.Outcomes[OutcomeKey("mesh-x", "src2:f1", "ace", "run")]; got.Outcome != OutcomeRunning { + if got := back.Outcomes[OutcomeKey("mesh-x", "src2:f1", "ace", StepRunning)]; got.Outcome != OutcomeRunning { t.Fatalf("outcomes %+v", back.Outcomes) } }) + // W11.9: a record ends when every present node runs its entries, naming the nodes away, and says record-ended; a node + // away that wakes adds its outcomes and is no longer shown away. + t.Run("it ends once every present node runs its entries, and a node away that wakes adds its outcomes", func(t *testing.T) { + _, rs, r, c := setup(t) + var ended []recordEnded + rs.Say = func(subject string, body any, id string) error { + if subject == meshdelivery.SubjectRecordEnded && id == "record-ended."+r.Instance { + ended = append(ended, body.(recordEnded)) + } + return nil + } + if err := rs.AddEntry(r, RecordEntry{Module: "mesh-x", Fingerprint: "src2:f1", Node: "g14"}); err != nil { + t.Fatal(err) + } + if err := rs.AddEntry(r, RecordEntry{Module: "mesh-x", Fingerprint: "src2:f1"}); err != nil { + t.Fatal(err) // the catalog half's entry, which no node runs + } + running := func(node string) stdio.Envelope { + return stdio.Envelope{Key: "node-engine.running." + node + ".mesh-x", Node: node, + Body: json.RawMessage(`{"module":"mesh-x","fingerprint":"src2:f1"}`), Headers: map[string]string{"x-event-id": "r-" + node}} + } + if err := c.Hear(running("ace")); err != nil { + t.Fatal(err) + } + if back, _ := rs.Read(r.Instance); back.Ended != nil || len(ended) != 0 { + t.Fatalf("a record whose difference is not known ended: %+v", back.Ended) + } + kept, _ := rs.Kept() // the records as the consumer holds them + r = kept[0] + if err := rs.SetTrace(r, RecordTrace{From: "assignment-8", To: "assignment-9"}); err != nil { + t.Fatal(err) + } + if done, err := rs.EndIfRun(r); err != nil || done || allRun(r) != "waiting for mesh-x on g14" { + t.Fatalf("ended %v (%v) while g14 runs nothing: %q", done, err, allRun(r)) + } + if err := rs.NodeAway(r, "g14", at.Add(-time.Hour)); err != nil { + t.Fatal(err) + } + back, _ := rs.Read(r.Instance) + if back.Ended == nil || back.Ended.Why != WhyAllRun || len(back.Ended.Away) != 1 || back.Ended.Away[0].Node != "g14" || + !back.Ended.Away[0].Since.Equal(at.Add(-time.Hour)) { + t.Fatalf("ended %+v", back.Ended) + } + if len(ended) != 1 || ended[0].Milestone != "47/G" || len(ended[0].Away) != 1 || ended[0].Away[0].Node != "g14" { + t.Fatalf("record-ended said %+v", ended) + } + if err := c.Hear(running("g14")); err != nil { + t.Fatal(err) + } + back, _ = rs.Read(r.Instance) + if got := back.Outcomes[OutcomeKey("mesh-x", "src2:f1", "g14", StepRunning)]; got.Outcome != OutcomeRunning || len(back.Away) != 0 { + t.Fatalf("the node that woke: outcomes %+v, away %+v", back.Outcomes, back.Away) + } + if back.Ended == nil || len(back.Ended.Away) != 1 { + t.Fatalf("how it ended is not kept: %+v", back.Ended) + } + }) 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") @@ -449,3 +568,53 @@ func TestTheSwitchCarriesTheGroupsInFlightIntoRecordsOnce(t *testing.T) { t.Fatalf("the take-in of the group's merge is not its record's outcome: %+v", back.Outcomes) } } + +// W11.9 (design 47 *The delivery record*): the gate says heads-replaced when a candidate's heads were replaced or its pull +// request closed, keyed as the candidate's revisions are (the heads' hash), so the indexer and the assigner drop its graphs. +func TestACandidatesHeadsReplacedOrClosedAreSaid(t *testing.T) { + a, b, c := strings.Repeat("a", 40), strings.Repeat("b", 40), strings.Repeat("c", 40) + setup := func(t *testing.T) (*world, *[]headsReplaced) { + w := newWorld(t) + var said []headsReplaced + w.h.SaySeat = func(subject string, body any, id string) error { + if subject != meshdelivery.SubjectHeadsReplaced { + t.Errorf("said on %s", subject) + } + e := body.(headsReplaced) + if id != "heads-replaced."+e.Group+"."+e.Heads { + t.Errorf("id %s for %+v", id, e) + } + said = append(said, e) + return nil + } + w.h.PullUpdated(pr("novox/app", 7, a, "feat/x", "src/a.go")) + w.h.PullUpdated(pr("novox/lib", 3, b, "feat/x", "src/b.go")) + w.settleAll() + return w, &said + } + hashOf := func(heads ...mdc.Head) string { + h, err := mdc.HeadsHash(heads) + if err != nil { + t.Fatal(err) + } + return h + } + group := hashOf(mdc.Head{Repository: "novox/app", Number: 7, Head: a}, mdc.Head{Repository: "novox/lib", Number: 3, Head: b}) + t.Run("a member's new head replaces the group's candidate", func(t *testing.T) { + w, said := setup(t) + w.h.PullUpdated(pr("novox/app", 7, c, "feat/x", "src/a.go")) + w.settleAll() + if len(*said) != 1 || (*said)[0].Group != "feat/x" || (*said)[0].Heads != group || (*said)[0].Head != c || + (*said)[0].Closed || (*said)[0].Repository != "novox/app" || (*said)[0].Number != 7 { + t.Fatalf("said %+v (the candidate %s)", *said, group) + } + }) + t.Run("a member's pull request closed replaces the group's candidate", func(t *testing.T) { + w, said := setup(t) + w.h.PullClosed(pr("novox/lib", 3, b, "feat/x")) + w.settleAll() + if len(*said) != 1 || (*said)[0].Heads != group || !(*said)[0].Closed || (*said)[0].Head != "" || (*said)[0].Number != 3 { + t.Fatalf("said %+v (the candidate %s)", *said, group) + } + }) +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordevents.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordevents.go new file mode 100644 index 000000000..5e2d7978a --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordevents.go @@ -0,0 +1,293 @@ +package main + +// What a delivery record says on the seat `mesh-delivery` (design 47 *The delivery record*): `pinned` and `freed` for the +// graph revisions it starts and stops pointing at, `heads-replaced` for a candidate whose heads were replaced or whose pull +// request closed, and `record-ended` once every present node runs its entries. Only the gate's account publishes them +// (ADR 0363 §2); each is said under an id of what it says, so said again it is kept once, and a say that failed is said +// again at the next write of the record. + +import ( + "fmt" + "sort" + "strconv" + "strings" + "time" + + mdc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" +) + +// The halves of the revision view a record pins (design 47 *Two graphs, one source of truth*). +const ( + HalfCatalog = "catalog" + HalfAssignment = "assignment" +) + +// StepRunning is the step whose outcome OutcomeRunning says an entry runs healthy on its node: from the node-engine seat's +// `running` event, never from a have read across the seam (ADR 0596). +const StepRunning = "running" + +// WhyAllRun is why a record ends by itself. +const WhyAllRun = "every present node runs its entries" + +// SeatSay says one event of the seat mesh-delivery on its subject, under an id the bus keeps once (stdio.SeatPublish). +type SeatSay func(subject string, body any, id string) error + +func (rs *Records) say(subject string, body any, id string) error { + if rs.Say == nil { + return nil + } + return rs.Say(subject, body, id) +} + +// pinKey is a pinned revision's key in the record: its half and its number, so the two halves' numbers never meet. +func pinKey(half string, revision int64) string { return half + "-" + strconv.FormatInt(revision, 10) } + +// number is the revision's number on its half's seat. +func (p PinnedRevision) number() (int64, bool) { + n, err := strconv.ParseInt(strings.TrimPrefix(p.Revision, p.Half+"-"), 10, 64) + return n, err == nil && strings.HasPrefix(p.Revision, p.Half+"-") +} + +// pinsOf is the seat's Pins of a record for these revisions. +func pinsOf(r *Record, pins []PinnedRevision, why string) meshdelivery.Pins { + out := meshdelivery.Pins{Record: r.Instance, Milestone: r.Milestone, Why: why} + for _, p := range pins { + n, ok := p.number() + if !ok { + continue + } + switch p.Half { + case HalfCatalog: + out.Catalog = append(out.Catalog, n) + case HalfAssignment: + out.Assignment = append(out.Assignment, n) + } + } + return out +} + +func checkHalf(half string) error { + if half != HalfCatalog && half != HalfAssignment { + return fmt.Errorf("a record pins a catalog or an assignment graph revision, not a %q one", half) + } + return nil +} + +// Pin writes that the record points at a graph revision, unless it does already, and says `pinned` (again, when written +// before: the id keeps it once). +func (rs *Records) Pin(r *Record, half string, revision int64) error { + if err := checkHalf(half); err != nil { + return err + } + p := PinnedRevision{Revision: pinKey(half, revision), Half: half} + if _, ok := r.Pinned[p.Revision]; !ok { + seq, err := rs.Ledger.Apply(r.Instance, OpPin, p) + if err != nil { + return err + } + if r.Pinned == nil { + r.Pinned = map[string]PinnedRevision{} + } + r.Pinned[p.Revision], r.Sequence = p, seq + } + return rs.say(meshdelivery.SubjectPinned, pinsOf(r, []PinnedRevision{p}, ""), "pinned."+r.Instance+"."+p.Revision) +} + +// Free writes that the record no longer points at a graph revision and says `freed` with why; a revision it does not +// point at frees nothing and says nothing. +func (rs *Records) Free(r *Record, half string, revision int64, why string) error { + if err := checkHalf(half); err != nil { + return err + } + p, ok := r.Pinned[pinKey(half, revision)] + if !ok { + return nil + } + if err := rs.say(meshdelivery.SubjectFreed, pinsOf(r, []PinnedRevision{p}, why), "freed."+r.Instance+"."+p.Revision); err != nil { + return err + } + seq, err := rs.Ledger.Apply(r.Instance, OpFree, map[string]any{"revision": p.Revision}) + if err != nil { + return err + } + delete(r.Pinned, p.Revision) + r.Sequence = seq + return nil +} + +// sayFreedAll says `freed` for every revision a record pins, before the record is removed: said first, so a removal never +// leaves a revision pinned by nobody that the indexer or the assigner still keeps. +func (rs *Records) sayFreedAll(r *Record, why string) error { + pins := sortedPins(r) + if len(pins) == 0 { + return nil + } + return rs.say(meshdelivery.SubjectFreed, pinsOf(r, pins, why), "freed."+r.Instance) +} + +// NodeAway writes a node away since when, unless the record holds it already, and ends the record when every other node +// runs its entries: a node away holds nothing (ADR 0603 §7). +func (rs *Records) NodeAway(r *Record, node string, since time.Time) error { + if node == "" { + return fmt.Errorf("a node away is named") + } + if _, ok := r.Away[node]; !ok { + a := NodeAway{Node: node, Since: since.UTC()} + seq, err := rs.Ledger.Apply(r.Instance, OpAway, a) + if err != nil { + return err + } + if r.Away == nil { + r.Away = map[string]NodeAway{} + } + r.Away[node], r.Sequence = a, seq + } + _, err := rs.EndIfRun(r) + return err +} + +// NodeBack writes that a node away is heard again: the record no longer shows it away, also once it ended. +func (rs *Records) NodeBack(r *Record, node string) error { + if _, ok := r.Away[node]; !ok { + return nil + } + seq, err := rs.Ledger.Apply(r.Instance, OpBack, map[string]any{"node": node}) + if err != nil { + return err + } + delete(r.Away, node) + r.Sequence = seq + return nil +} + +// allRun is why a record does not end yet, or "" when every present node runs its entries: the difference it delivers is +// known (the trace names the assignment graph revision it moved to), and every entry on a node not away has its running +// healthy outcome. An entry of the catalog half names no node and is run by none. +func allRun(r *Record) string { + if r.Trace == nil || r.Trace.To == "" { + return "its difference is not known yet" + } + var waiting []string + for _, e := range r.Entries { + if e.Node == "" { + continue + } + if _, away := r.Away[e.Node]; away { + continue + } + if o, ok := r.Outcomes[OutcomeKey(e.Module, e.Fingerprint, e.Node, StepRunning)]; !ok || o.Outcome != OutcomeRunning { + waiting = append(waiting, e.Module+" on "+e.Node) + } + } + if len(waiting) > 0 { + sort.Strings(waiting) + return "waiting for " + strings.Join(waiting, ", ") + } + return "" +} + +// awayList is the record's nodes away, by node. +func awayList(r *Record) []NodeAway { + out := make([]NodeAway, 0, len(r.Away)) + for _, a := range r.Away { + out = append(out, a) + } + sort.Slice(out, func(i, j int) bool { return out[i].Node < out[j].Node }) + return out +} + +// EndIfRun ends the record when every present node runs its entries, naming the nodes away with since when, and says +// `record-ended`. Ended says it ended now; a record ended before is said again (the id keeps it once). +func (rs *Records) EndIfRun(r *Record) (ended bool, err error) { + if r.Ended == nil { + if allRun(r) != "" { + return false, nil + } + if err := rs.end(r, WhyAllRun, ""); err != nil { + return false, err + } + ended = true + } + return ended, rs.sayEnded(r) +} + +// End ends a record by hand, with the reason and who (gate_record_end). +func (rs *Records) End(r *Record, why, by string) error { + if r.Ended != nil { + return rs.sayEnded(r) + } + if strings.TrimSpace(why) == "" { + return fmt.Errorf("a record ended by hand says why") + } + if err := rs.end(r, why, by); err != nil { + return err + } + return rs.sayEnded(r) +} + +func (rs *Records) end(r *Record, why, by string) error { + e := RecordEnded{At: rs.now(), Why: why, By: by, Away: awayList(r)} + seq, err := rs.Ledger.Apply(r.Instance, OpEnd, map[string]any{"ended": e}) + if err != nil { + return err + } + r.Ended, r.Sequence = &e, seq + rs.keep(r) + return nil +} + +// recordEnded is the seat's `record-ended` body. +type recordEnded struct { + Record string `json:"record"` + Milestone string `json:"milestone"` + ForgeMade bool `json:"forge-made,omitempty"` + Away []NodeAway `json:"away,omitempty"` + At time.Time `json:"at"` +} + +func (rs *Records) sayEnded(r *Record) error { + if r.Ended == nil { + return nil + } + return rs.say(meshdelivery.SubjectRecordEnded, recordEnded{Record: r.Instance, Milestone: r.Milestone, + ForgeMade: r.ForgeMade, Away: r.Ended.Away, At: r.Ended.At}, "record-ended."+r.Instance) +} + +// headsReplaced is the seat's `heads-replaced` body: a candidate's heads replaced or its pull request closed. +type headsReplaced struct { + Group string `json:"group"` + Heads string `json:"heads"` + Repository string `json:"repository"` + Number int `json:"number"` + Head string `json:"head,omitempty"` + Closed bool `json:"closed,omitempty"` + At time.Time `json:"at"` +} + +// sayHeadsReplaced says that the candidate of these members' heads means nothing any more, since one member's head moved +// to head or (closed) its pull request closed; the heads are keyed as the candidate's revisions are (HeadsHash). Under the +// lock; the say itself is the work's, outside it. +func (h *Holder) sayHeadsReplaced(group string, members []*Delivery, moved *Delivery, head string, closed bool) { + if h.SaySeat == nil || len(members) == 0 { + return + } + hash, err := mdc.HeadsHash(headsOf(members)) + if err != nil { + h.refuse(fmt.Errorf("the candidate of %s cannot be said replaced: %v", group, err)) + return + } + e := headsReplaced{Group: group, Heads: hash, Repository: strings.ToLower(moved.Repository), Number: moved.Number, + Closed: closed, At: h.Now().UTC()} + if !closed { + e.Head = strings.ToLower(head) + } + say := h.SaySeat + id := "heads-replaced." + group + "." + hash + h.cancels = append(h.cancels, func() { + if err := say(meshdelivery.SubjectHeadsReplaced, e, id); err != nil { + h.Logf("[mesh-delivery] the candidate %s of %s is not said replaced: %v", hash, group, err) + } + }) + h.Poke() +}