The delivery record says pinned, freed, heads-replaced and record-ended; it ends once every present node runs its entries (milestone 47/G, W11.9)
A node away is named with since when and holds nothing; a node away that wakes adds its outcomes and is no longer shown away. freed is said before a record past its retention is removed, so no revision stays pinned by nobody. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ (Joined from 52b629c6.) Builds: design 47 G §147, §149, §153 (GATE3) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
Reference in New Issue
Block a user