| new file mode 100644 |
| index 000000000..129546f60 |
| --- /dev/null |
| +++ b/internal/ops/admissionmodel/model.go |
| @@ -0,0 +1,931 @@ |
| +// Package admissionmodel is plan step 0's second spike for the replicator's |
| +// authority design (docs/superpowers/specs/2026-10-03-replicator-pinned-roots-design.md, |
| +// Part C): the admission's transition table, the two kinds of hold, the |
| +// pending-discovery latch and its replay, construction backfill, ancestry |
| +// reach, acknowledgement, helper death, the producer fence and process |
| +// recovery, as one executable model with no SQLite and no clock. The tests |
| +// drive it through the sequences the design names and through randomised |
| +// schedules, and check the invariants the design claims (I2) against a truth |
| +// the model keeps separately from the ledger it simulates. Nothing here is |
| +// wired; it exists to find the interleavings a prose table hides. |
| +package admissionmodel |
| + |
| +import ( |
| + "fmt" |
| + "math/rand" |
| + "sort" |
| +) |
| + |
| +// Time is minutes since an epoch; a Day is 1440 of them. Margin is the |
| +// design's one minute. |
| +type Time = int64 |
| + |
| +const ( |
| + Day Time = 1440 |
| + Margin Time = 1 |
| + unset Time = -1 |
| +) |
| + |
| +func dayOf(t Time) int64 { return t / Day } |
| + |
| +// Owner is an (account, app) pair. |
| +type Owner string |
| + |
| +// State is an admission's state in the design's table. |
| +type State int |
| + |
| +const ( |
| + Registered State = iota // in memory only |
| + Admitted // durable row |
| + TransportDone |
| + RecorderDone |
| + Settled |
| + Unissued |
| + Cancelled |
| + RecoveredUnknown |
| +) |
| + |
| +func (s State) String() string { |
| + return [...]string{"registered", "admitted", "transport done", "recorder done", "settled", "unissued", "cancelled", "recovered-unknown"}[s] |
| +} |
| + |
| +// Terminal reports whether no participant remains to complete. |
| +func (s State) Terminal() bool { return s >= Settled } |
| + |
| +// Admission is one row of `admissions`, plus the in-memory facts of its |
| +// registration and the truth the model keeps about what the helper did. |
| +type Admission struct { |
| + Token int |
| + Fleet *Fleet |
| + Owner Owner |
| + AdmittedAt Time |
| + State State |
| + Durable bool // the row exists |
| + Issued bool // the token was handed to the helper |
| + Refused bool // the gate's budget expired or the recheck refused |
| + |
| + writerRunning bool |
| + epoch int // the fleet's discovery epoch at registration |
| + |
| + TransportDoneAt Time |
| + RecorderStartedAt Time |
| + RecorderDoneAt Time |
| + SettledAt Time |
| + How string |
| + |
| + // truth, never read by the ledger side |
| + SentAt Time // when the helper began the transport |
| + SentEndAt Time // when the transport truly ended (lost: helper death) |
| +} |
| + |
| +func (a *Admission) unsettled() bool { return !a.State.Terminal() } |
| + |
| +// Uncertainty is one row of `uncertainties` with the tokens it counted. |
| +type Uncertainty struct { |
| + ID int |
| + Fleet *Fleet |
| + From, To Time // To == unset while open |
| + DiscoveryAt Time |
| + Counted map[int]*Admission |
| +} |
| + |
| +func (u *Uncertainty) open() bool { return u.To == unset } |
| + |
| +// Latch is the pending discovery the ledger's failure overlay carries. |
| +type Latch struct { |
| + Fleet *Fleet |
| + From Time |
| + At Time |
| + Counted map[int]*Admission |
| +} |
| + |
| +// FState is a fleet row's state. |
| +type FState int |
| + |
| +const ( |
| + Open FState = iota |
| + Stale |
| + Closing |
| + Closed |
| +) |
| + |
| +// Fleet is one row of `fleets` plus its in-memory state. |
| +type Fleet struct { |
| + ID int |
| + Owners []Owner |
| + OpenedAt Time |
| + State FState |
| + Pred *Fleet |
| + Acked bool // this fleet has acknowledged its predecessor |
| + |
| + Admissions []*Admission |
| + Uncertainties []*Uncertainty |
| + Inherited []*Uncertainty |
| + |
| + // in memory |
| + pending map[int]*Admission |
| + epoch int |
| + inDiscovery bool |
| + latch *Latch |
| + lastCleanLook Time |
| + proof Time |
| + helperAlive bool |
| + helperStopped bool |
| + guardClosed bool |
| + |
| + // truth |
| + ChangedAt Time // the instant authority truly changed (unset if never) |
| + DiscoveredAt Time // when the ledger learned of it (durably or latched) |
| + DiedAt Time // helper death |
| +} |
| + |
| +// Model is the whole box: one agent process, its helpers, the ledger. |
| +type Model struct { |
| + rnd *rand.Rand |
| + Now Time |
| + Fleets []*Fleet |
| + Admissions []*Admission |
| + Holds map[int64]map[Owner]bool // durable |
| + Alive bool // the agent process |
| + Recovered bool // recovery ran since the last crash |
| + CrashedAt Time |
| + nextToken int |
| + nextUnc int |
| + Log []string |
| +} |
| + |
| +// New returns an empty box with a live agent. |
| +func New(seed int64) *Model { |
| + return &Model{rnd: rand.New(rand.NewSource(seed)), Holds: map[int64]map[Owner]bool{}, Alive: true, Recovered: true, CrashedAt: unset} |
| +} |
| + |
| +func (m *Model) logf(f string, args ...any) { |
| + m.Log = append(m.Log, fmt.Sprintf("%6d %s", m.Now, fmt.Sprintf(f, args...))) |
| +} |
| + |
| +// Advance moves the clock. |
| +func (m *Model) Advance(d Time) { m.Now += d } |
| + |
| +func (m *Model) hold(day int64, o Owner) { |
| + if m.Holds[day] == nil { |
| + m.Holds[day] = map[Owner]bool{} |
| + } |
| + m.Holds[day][o] = true |
| +} |
| + |
| +func (m *Model) holdInterval(from, to Time, owners []Owner) { |
| + for d := dayOf(from); d <= dayOf(to); d++ { |
| + for _, o := range owners { |
| + m.hold(d, o) |
| + } |
| + } |
| +} |
| + |
| +// unackAncestry is the design's walk: predecessors through fleets that have |
| +// not acknowledged; an acknowledged step ends it. |
| +func unackAncestry(g *Fleet) []*Fleet { |
| + var anc []*Fleet |
| + for cur := g; !cur.Acked && cur.Pred != nil; cur = cur.Pred { |
| + anc = append(anc, cur.Pred) |
| + } |
| + return anc |
| +} |
| + |
| +// effective is a fleet's effective set: own, inherited, and the own and |
| +// inherited sets of its unacknowledged ancestry. |
| +func effective(g *Fleet) []*Uncertainty { |
| + seen := map[int]bool{} |
| + var out []*Uncertainty |
| + add := func(us []*Uncertainty) { |
| + for _, u := range us { |
| + if !seen[u.ID] { |
| + seen[u.ID] = true |
| + out = append(out, u) |
| + } |
| + } |
| + } |
| + add(g.Uncertainties) |
| + add(g.Inherited) |
| + for _, a := range unackAncestry(g) { |
| + add(a.Uncertainties) |
| + add(a.Inherited) |
| + } |
| + return out |
| +} |
| + |
| +// reach is who an uncertainty of f reaches: f's owners and the owners of |
| +// every fleet whose unacknowledged ancestry contains f. |
| +func (m *Model) reach(f *Fleet) []Owner { |
| + set := map[Owner]bool{} |
| + for _, o := range f.Owners { |
| + set[o] = true |
| + } |
| + for _, g := range m.Fleets { |
| + for _, a := range unackAncestry(g) { |
| + if a == f { |
| + for _, o := range g.Owners { |
| + set[o] = true |
| + } |
| + } |
| + } |
| + } |
| + return ownersOf(set) |
| +} |
| + |
| +func ownersOf(set map[Owner]bool) []Owner { |
| + out := make([]Owner, 0, len(set)) |
| + for o := range set { |
| + out = append(out, o) |
| + } |
| + sort.Slice(out, func(i, j int) bool { return out[i] < out[j] }) |
| + return out |
| +} |
| + |
| +// Build constructs a fleet against the chain's tip, with construction |
| +// backfill: every interval its unacknowledged ancestry already holds. |
| +func (m *Model) Build(owners ...Owner) *Fleet { |
| + m.mustAlive("build") |
| + var pred *Fleet |
| + if n := len(m.Fleets); n > 0 { |
| + pred = m.Fleets[n-1] |
| + } |
| + f := &Fleet{ID: len(m.Fleets) + 1, Owners: owners, OpenedAt: m.Now, Pred: pred, pending: map[int]*Admission{}, |
| + lastCleanLook: m.Now, proof: m.Now, helperAlive: true, ChangedAt: unset, DiscoveredAt: unset, DiedAt: unset} |
| + m.Fleets = append(m.Fleets, f) |
| + for _, u := range effective(f) { // own set is empty; this is the ancestry's |
| + to := u.To |
| + if u.open() { |
| + to = m.Now |
| + } |
| + m.holdInterval(u.From, to, f.Owners) |
| + } |
| + m.logf("fleet %d built (owners %v, pred %v)", f.ID, owners, predID(pred)) |
| + return f |
| +} |
| + |
| +func (f *Fleet) owns(o Owner) bool { |
| + for _, x := range f.Owners { |
| + if x == o { |
| + return true |
| + } |
| + } |
| + return false |
| +} |
| + |
| +func predID(p *Fleet) any { |
| + if p == nil { |
| + return nil |
| + } |
| + return p.ID |
| +} |
| + |
| +// Look is a clean look at the fleet's files: it moves the lower bound. If |
| +// the authority has truly changed, the look is a discovery instead. |
| +func (m *Model) Look(f *Fleet) { |
| + m.mustAlive("look") |
| + if f.ChangedAt != unset && f.DiscoveredAt == unset { |
| + m.Discover(f, false) |
| + return |
| + } |
| + if f.State == Open { |
| + f.lastCleanLook = m.Now |
| + } |
| +} |
| + |
| +// Change is the truth: the fleet's authority changes now, unseen. |
| +func (m *Model) Change(f *Fleet) { |
| + if f.ChangedAt == unset { |
| + f.ChangedAt = m.Now |
| + } |
| +} |
| + |
| +// Register is the gate's first half: under the discovery mutex, proof then |
| +// registration in the pending set. It returns nil when refused. |
| +func (m *Model) Register(f *Fleet, o Owner) *Admission { |
| + m.mustAlive("register") |
| + if f.State != Open || f.inDiscovery || f.latch != nil || !f.helperAlive || f.helperStopped { |
| + return nil |
| + } |
| + if !f.owns(o) { |
| + panic("model: an admission for an owner the fleet does not have") |
| + } |
| + f.proof = m.Now // the gate's proof: the witness read unchanged now |
| + m.nextToken++ |
| + a := &Admission{Token: m.nextToken, Fleet: f, Owner: o, AdmittedAt: m.Now, State: Registered, writerRunning: true, epoch: f.epoch, |
| + TransportDoneAt: unset, RecorderStartedAt: unset, RecorderDoneAt: unset, SettledAt: unset, SentAt: unset, SentEndAt: unset} |
| + f.pending[a.Token] = a |
| + m.Admissions = append(m.Admissions, a) |
| + m.logf("T%d registered on fleet %d for %s", a.Token, f.ID, o) |
| + return a |
| +} |
| + |
| +// BudgetExpires: the gate refuses the request while the writer still runs. |
| +func (m *Model) BudgetExpires(a *Admission) { |
| + if a.State == Registered && a.writerRunning { |
| + a.Refused = true |
| + } |
| +} |
| + |
| +// WriterReturns resolves a registration: "commit", "fail", or "ambiguous" |
| +// (the writer cannot tell; the model decides whether the write landed and |
| +// the read-back, after the handle is closed, finds it). |
| +func (m *Model) WriterReturns(a *Admission, outcome string) { |
| + if !m.Alive || a.State != Registered || !a.writerRunning { |
| + return |
| + } |
| + a.writerRunning = false |
| + f := a.Fleet |
| + landed := outcome == "commit" || (outcome == "ambiguous" && m.rnd.Intn(2) == 0) |
| + if outcome == "ambiguous" { |
| + a.Refused = true // the gate could not answer within its budget |
| + } |
| + if !landed { |
| + // conclusive cancellation: the writer has returned (or its handle is |
| + // closed and a fresh read found no row) |
| + a.State = Cancelled |
| + a.SettledAt = m.Now |
| + a.How = "cancelled" |
| + delete(f.pending, a.Token) |
| + m.logf("T%d cancelled", a.Token) |
| + m.emptySettlement(a) |
| + return |
| + } |
| + a.Durable = true |
| + a.State = Admitted |
| + f.Admissions = append(f.Admissions, a) |
| + delete(f.pending, a.Token) // under the mutex: the row now carries it |
| + // the recheck under the mutex (the epoch test is redundant with the |
| + // stale state here, since every discovery stales the fleet; it is kept |
| + // because the design names it) |
| + if a.Refused || f.State != Open || f.inDiscovery || f.epoch != a.epoch || f.latch != nil { |
| + m.terminal(a, Unissued, "unissued") |
| + m.logf("T%d unissued", a.Token) |
| + m.emptySettlement(a) |
| + return |
| + } |
| + a.Issued = true |
| + a.SentAt = m.Now |
| + m.logf("T%d issued", a.Token) |
| +} |
| + |
| +func (m *Model) terminal(a *Admission, s State, how string) { |
| + a.State = s |
| + a.How = how |
| + if a.TransportDoneAt == unset { |
| + a.TransportDoneAt = m.Now |
| + } |
| + if a.RecorderDoneAt == unset { |
| + a.RecorderDoneAt = m.Now |
| + } |
| + a.SettledAt = m.Now |
| +} |
| + |
| +// emptySettlement tells every discovery that counted a registration that it |
| +// ended with an empty interval, so an uncertainty waiting only on it closes. |
| +func (m *Model) emptySettlement(a *Admission) { |
| + for _, u := range a.Fleet.Uncertainties { |
| + if u.open() && u.Counted[a.Token] != nil { |
| + m.maybeClose(u, m.Now) |
| + } |
| + } |
| +} |
| + |
| +func (m *Model) openUncertainty(f *Fleet) *Uncertainty { |
| + for _, u := range f.Uncertainties { |
| + if u.open() { |
| + return u |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +func countedUnsettled(u *Uncertainty) bool { |
| + for _, a := range u.Counted { |
| + if a.unsettled() { |
| + return true |
| + } |
| + } |
| + return false |
| +} |
| + |
| +func (m *Model) maybeClose(u *Uncertainty, settlement Time) { |
| + if countedUnsettled(u) { |
| + return |
| + } |
| + to := u.DiscoveryAt |
| + if settlement > to { |
| + to = settlement |
| + } |
| + u.To = to + Margin |
| + m.holdInterval(u.From, u.To, m.reach(u.Fleet)) |
| + m.logf("U%d closed [%d,%d]", u.ID, u.From, u.To) |
| +} |
| + |
| +// Discover is the change found: under the discovery mutex, the lower bound |
| +// and the counted set; then the transaction, which may fail and latch. |
| +func (m *Model) Discover(f *Fleet, fail bool) { |
| + m.mustAlive("discover") |
| + if f.State != Open { |
| + return |
| + } |
| + f.inDiscovery = true |
| + f.epoch++ |
| + if f.ChangedAt == unset { |
| + f.ChangedAt = m.Now |
| + } |
| + f.DiscoveredAt = m.Now |
| + counted := map[int]*Admission{} |
| + from := f.proof |
| + if f.lastCleanLook < from { |
| + from = f.lastCleanLook |
| + } |
| + for _, a := range f.Admissions { |
| + if a.unsettled() { |
| + counted[a.Token] = a |
| + if a.AdmittedAt < from { |
| + from = a.AdmittedAt |
| + } |
| + } |
| + } |
| + for _, a := range f.pending { |
| + counted[a.Token] = a |
| + if a.AdmittedAt < from { |
| + from = a.AdmittedAt |
| + } |
| + } |
| + from -= Margin |
| + f.State = Stale |
| + f.inDiscovery = false |
| + if fail { |
| + f.latch = &Latch{Fleet: f, From: from, At: m.Now, Counted: counted} |
| + m.logf("fleet %d discovery latched (from %d, %d counted)", f.ID, from, len(counted)) |
| + return |
| + } |
| + m.nextUnc++ |
| + u := &Uncertainty{ID: m.nextUnc, Fleet: f, From: from, To: unset, DiscoveryAt: m.Now, Counted: counted} |
| + f.Uncertainties = append(f.Uncertainties, u) |
| + m.holdInterval(from, m.Now, m.reach(f)) |
| + m.logf("fleet %d discovered: U%d open from %d (%d counted)", f.ID, u.ID, from, len(counted)) |
| + m.maybeClose(u, m.Now) |
| +} |
| + |
| +// Replay is the failed discovery's one transaction. |
| +func (m *Model) Replay(f *Fleet) { |
| + m.mustAlive("replay") |
| + l := f.latch |
| + if l == nil { |
| + return |
| + } |
| + m.nextUnc++ |
| + u := &Uncertainty{ID: m.nextUnc, Fleet: f, From: l.From, To: unset, DiscoveryAt: l.At, Counted: l.Counted} |
| + f.Uncertainties = append(f.Uncertainties, u) |
| + reached := m.reach(f) |
| + m.holdInterval(l.From, l.At, reached) |
| + for _, a := range l.Counted { |
| + if a.State == Settled { |
| + m.holdInterval(a.AdmittedAt, a.SettledAt, reached) |
| + } |
| + } |
| + if !countedUnsettled(u) { |
| + u.To = m.Now + Margin |
| + m.holdInterval(u.From, u.To, reached) |
| + } |
| + f.latch = nil // the overlay clears after the commit |
| + m.logf("fleet %d replayed: U%d from %d to %d", f.ID, u.ID, u.From, u.To) |
| +} |
| + |
| +// TransportDone is the SDK middleware's finalize step. |
| +func (m *Model) TransportDone(a *Admission) { |
| + if !a.Issued || a.TransportDoneAt != unset { |
| + return |
| + } |
| + a.TransportDoneAt = m.Now |
| + a.SentEndAt = m.Now |
| + if a.State == Admitted { |
| + a.State = TransportDone |
| + } else if a.State == RecorderDone { |
| + m.settle(a) |
| + } |
| +} |
| + |
| +// RecorderStart is the helper's Start message reaching the agent. |
| +func (m *Model) RecorderStart(a *Admission) bool { |
| + if !m.Alive || !a.Issued || a.RecorderStartedAt != unset || a.RecorderDoneAt != unset || a.Fleet.guardClosed { |
| + return false |
| + } |
| + a.RecorderStartedAt = m.Now |
| + return true |
| +} |
| + |
| +// RecorderDone is Finish, Abandon, or unrecorded (Start failed). |
| +func (m *Model) RecorderDone(a *Admission, how string) { |
| + if !m.Alive || !a.Issued || a.RecorderDoneAt != unset { |
| + return |
| + } |
| + if how != "unrecorded" && a.RecorderStartedAt == unset { |
| + return |
| + } |
| + a.RecorderDoneAt = m.Now |
| + a.How = how |
| + if a.State == Admitted { |
| + a.State = RecorderDone |
| + } else if a.State == TransportDone { |
| + m.settle(a) |
| + } |
| +} |
| + |
| +func (m *Model) settle(a *Admission) { |
| + a.State = Settled |
| + a.SettledAt = a.TransportDoneAt |
| + if a.RecorderDoneAt > a.SettledAt { |
| + a.SettledAt = a.RecorderDoneAt |
| + } |
| + if a.SettledAt < m.Now { |
| + a.SettledAt = m.Now |
| + } |
| + f := a.Fleet |
| + if u := m.openUncertainty(f); u != nil { |
| + m.holdInterval(a.AdmittedAt, a.SettledAt, m.reach(f)) |
| + m.maybeClose(u, a.SettledAt) |
| + } |
| + m.logf("T%d settled", a.Token) |
| +} |
| + |
| +// HelperDies: the helper's death proven by waitpid. It is a change; the |
| +// guard closes, the socket drains, transports are lost and never-registered |
| +// recorders are unrecorded. Registered workers are kept. |
| +func (m *Model) HelperDies(f *Fleet, discoveryFails bool) { |
| + if !f.helperAlive { |
| + return |
| + } |
| + f.helperAlive = false |
| + f.DiedAt = m.Now |
| + if !m.Alive { |
| + return // the agent is gone; recovery handles it |
| + } |
| + m.Change(f) |
| + m.Discover(f, discoveryFails) |
| + f.guardClosed = true |
| + for _, a := range f.Admissions { |
| + if !a.Issued || a.State.Terminal() { |
| + continue |
| + } |
| + if a.TransportDoneAt == unset { |
| + a.TransportDoneAt = m.Now |
| + a.SentEndAt = m.Now |
| + if a.State == Admitted { |
| + a.State = TransportDone |
| + } |
| + } |
| + if a.RecorderDoneAt == unset && a.RecorderStartedAt == unset { |
| + a.RecorderDoneAt = m.Now |
| + a.How = "unrecorded" |
| + } |
| + if a.TransportDoneAt != unset && a.RecorderDoneAt != unset { |
| + m.settle(a) |
| + } |
| + } |
| +} |
| + |
| +// StopHelper is closeNow's clean path: only when no transport is in flight. |
| +func (m *Model) StopHelper(f *Fleet) bool { |
| + if !f.helperAlive { |
| + return false |
| + } |
| + for _, a := range f.Admissions { |
| + if a.Issued && a.TransportDoneAt == unset { |
| + return false |
| + } |
| + } |
| + m.Look(f) |
| + f.helperAlive = false |
| + f.helperStopped = true |
| + f.guardClosed = true |
| + for _, a := range f.Admissions { // Start can no longer arrive |
| + if a.Issued && !a.State.Terminal() && a.RecorderDoneAt == unset && a.RecorderStartedAt == unset { |
| + a.RecorderDoneAt = m.Now |
| + a.How = "unrecorded" |
| + if a.TransportDoneAt != unset { |
| + m.settle(a) |
| + } |
| + } |
| + } |
| + if f.State != Closed { |
| + f.State = Closing |
| + } |
| + return true |
| +} |
| + |
| +// Close marks the fleet closed when every admission is terminal and no |
| +// discovery is uncommitted; the closure carries a pending replay. |
| +func (m *Model) Close(f *Fleet) bool { |
| + m.mustAlive("close") |
| + if f.helperAlive || f.State == Closed { |
| + return false |
| + } |
| + for _, a := range f.Admissions { |
| + if a.unsettled() { |
| + return false |
| + } |
| + } |
| + if len(f.pending) > 0 { |
| + return false |
| + } |
| + if f.latch != nil { |
| + m.Replay(f) |
| + } |
| + f.State = Closed |
| + m.logf("fleet %d closed", f.ID) |
| + return true |
| +} |
| + |
| +// Acknowledge is root-first and copies the predecessor's effective set. |
| +func (m *Model) Acknowledge(g *Fleet) bool { |
| + m.mustAlive("acknowledge") |
| + p := g.Pred |
| + if p == nil || g.Acked || p.State != Closed || !(p.Acked || p.Pred == nil) { |
| + return false |
| + } |
| + for _, u := range effective(p) { |
| + g.Inherited = append(g.Inherited, u) |
| + if !u.open() { |
| + m.holdInterval(u.From, u.To, g.Owners) |
| + } |
| + } |
| + g.Acked = true |
| + m.logf("fleet %d acknowledged fleet %d", g.ID, p.ID) |
| + return true |
| +} |
| + |
| +// Crash kills the agent. In-memory state is lost; a registration whose |
| +// writer was running may or may not have landed; helpers may survive with |
| +// the fence. |
| +func (m *Model) Crash(helperSurvives func(*Fleet) bool) { |
| + if !m.Alive { |
| + return |
| + } |
| + m.Alive = false |
| + m.Recovered = false |
| + m.CrashedAt = m.Now |
| + for _, f := range m.Fleets { |
| + for tok, a := range f.pending { |
| + a.writerRunning = false |
| + if m.rnd.Intn(2) == 0 { // the commit landed before the crash |
| + a.Durable = true |
| + a.State = Admitted |
| + f.Admissions = append(f.Admissions, a) |
| + } else { |
| + a.State = Cancelled |
| + a.How = "lost-with-process" |
| + a.SettledAt = m.Now |
| + } |
| + delete(f.pending, tok) |
| + } |
| + f.latch = nil |
| + f.inDiscovery = false |
| + f.guardClosed = true |
| + if f.helperAlive && !helperSurvives(f) { |
| + f.helperAlive = false |
| + f.DiedAt = m.Now |
| + for _, a := range f.Admissions { |
| + if a.Issued && a.SentEndAt == unset { |
| + a.SentEndAt = m.Now |
| + } |
| + } |
| + } |
| + } |
| + m.logf("agent crashed") |
| +} |
| + |
| +// SurvivingHelperTransport: a helper that outlived its agent still finishes |
| +// a transport it had in hand. |
| +func (m *Model) SurvivingHelperTransport(a *Admission) { |
| + if m.Alive || !a.Fleet.helperAlive || !a.Issued || a.TransportDoneAt != unset { |
| + return |
| + } |
| + a.SentEndAt = m.Now // the truth; the row is not written, the agent is dead |
| +} |
| + |
| +// SurvivingHelperDies ends a helper the crash left alive. |
| +func (m *Model) SurvivingHelperDies(f *Fleet) { |
| + if f.helperAlive { |
| + f.helperAlive = false |
| + f.DiedAt = m.Now |
| + for _, a := range f.Admissions { |
| + if a.Issued && a.SentEndAt == unset { |
| + a.SentEndAt = m.Now |
| + } |
| + } |
| + } |
| +} |
| + |
| +// FenceFree reports whether every old producer is gone. |
| +func (m *Model) FenceFree() bool { |
| + for _, f := range m.Fleets { |
| + if f.helperAlive { |
| + return false |
| + } |
| + } |
| + return true |
| +} |
| + |
| +// Recover is the new process's first transaction, holding the fence. |
| +func (m *Model) Recover() bool { |
| + if m.Alive || !m.FenceFree() { |
| + return false |
| + } |
| + m.Alive = true |
| + m.Recovered = true |
| + for _, f := range m.Fleets { |
| + for _, a := range f.Admissions { |
| + if a.unsettled() { |
| + m.terminal(a, RecoveredUnknown, "recovered-unknown") |
| + m.nextUnc++ |
| + u := &Uncertainty{ID: m.nextUnc, Fleet: f, From: a.AdmittedAt - Margin, To: m.Now + Margin, DiscoveryAt: m.Now, Counted: map[int]*Admission{}} |
| + f.Uncertainties = append(f.Uncertainties, u) |
| + m.holdInterval(u.From, u.To, append(m.reach(f), a.Owner)) |
| + m.logf("T%d recovered-unknown, U%d", a.Token, u.ID) |
| + } |
| + } |
| + if f.State != Closed { |
| + if u := m.openUncertainty(f); u != nil { |
| + u.To = m.Now + Margin |
| + m.holdInterval(u.From, u.To, m.reach(f)) |
| + } else { |
| + from := f.OpenedAt |
| + for _, a := range f.Admissions { |
| + if a.AdmittedAt < from { |
| + from = a.AdmittedAt |
| + } |
| + } |
| + m.nextUnc++ |
| + u := &Uncertainty{ID: m.nextUnc, Fleet: f, From: from - Margin, To: m.Now + Margin, DiscoveryAt: m.Now, Counted: map[int]*Admission{}} |
| + f.Uncertainties = append(f.Uncertainties, u) |
| + m.holdInterval(u.From, u.To, m.reach(f)) |
| + } |
| + f.State = Closed |
| + f.helperAlive = false |
| + f.DiscoveredAt = m.Now |
| + } |
| + } |
| + // root-first acknowledgement of every fleet left unacknowledged |
| + for _, f := range m.Fleets { |
| + if f.Pred != nil && !f.Acked { |
| + m.Acknowledge(f) |
| + } |
| + } |
| + m.logf("recovered") |
| + return true |
| +} |
| + |
| +func (m *Model) mustAlive(op string) { |
| + if !m.Alive { |
| + panic("model: " + op + " while the agent is dead") |
| + } |
| +} |
| + |
| +// Complete is the publisher's answer for an owner's day: no hold, no |
| +// barrier, no pending discovery reaching the owner. It is only asked of a |
| +// live agent. |
| +func (m *Model) Complete(o Owner, day int64) bool { |
| + if !m.Alive { |
| + panic("model: publication asked of a dead agent") |
| + } |
| + if m.Holds[day][o] { |
| + return false |
| + } |
| + for _, a := range m.Admissions { |
| + if a.Durable && a.Issued && a.unsettled() && a.Owner == o && dayOf(a.AdmittedAt) <= day { |
| + return false |
| + } |
| + } |
| + for _, f := range m.Fleets { |
| + if f.latch != nil && dayOf(f.latch.From) <= day { |
| + for _, r := range m.reach(f) { |
| + if r == o { |
| + return false |
| + } |
| + } |
| + } |
| + } |
| + return true |
| +} |
| + |
| +// Check is the invariant pass the tests run after every step: a Violation |
| +// is returned for the first failure found. |
| +func (m *Model) Check() error { |
| + if !m.Alive { |
| + return nil |
| + } |
| + today := dayOf(m.Now) |
| + // barrier: an issued, unsettled admission bars its owner from its day on |
| + for _, a := range m.Admissions { |
| + if a.Durable && a.Issued && a.unsettled() { |
| + for d := dayOf(a.AdmittedAt); d <= today; d++ { |
| + if m.Complete(a.Owner, d) { |
| + return fmt.Errorf("barrier: T%d unsettled but %s day %d complete", a.Token, a.Owner, d) |
| + } |
| + } |
| + } |
| + } |
| + // hazard: a transport that overlapped a true change, once the ledger |
| + // knows of the change, has its days held for its owner |
| + for _, a := range m.Admissions { |
| + if !a.Issued || a.SentAt == unset { |
| + continue |
| + } |
| + f := a.Fleet |
| + known := f.DiscoveredAt != unset || (m.Recovered && m.CrashedAt != unset && a.AdmittedAt <= m.CrashedAt) |
| + if !known { |
| + continue |
| + } |
| + end := a.SentEndAt |
| + if end == unset && a.State.Terminal() { |
| + end = a.SettledAt |
| + } |
| + if end == unset { |
| + end = m.Now |
| + } |
| + hazard := false |
| + if f.ChangedAt != unset && end >= f.ChangedAt { |
| + hazard = true |
| + } |
| + if m.CrashedAt != unset && a.AdmittedAt <= m.CrashedAt && (a.SettledAt == unset || a.SettledAt >= m.CrashedAt) { |
| + hazard = true |
| + } |
| + if !hazard { |
| + continue |
| + } |
| + for d := dayOf(a.SentAt); d <= dayOf(end); d++ { |
| + if m.Complete(a.Owner, d) { |
| + return fmt.Errorf("hazard: T%d (sent %d..%d, change %d) but %s day %d complete", a.Token, a.SentAt, end, f.ChangedAt, a.Owner, d) |
| + } |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +// CheckQuiescent is the end-state pass: every fleet's owners hold every |
| +// interval of every uncertainty on its whole chain, and days nobody has a |
| +// reason to hold are complete. |
| +func (m *Model) CheckQuiescent() error { |
| + if err := m.Check(); err != nil { |
| + return err |
| + } |
| + for _, g := range m.Fleets { |
| + for p := g; p != nil; p = p.Pred { |
| + for _, u := range p.Uncertainties { |
| + if u.open() { |
| + return fmt.Errorf("U%d still open at quiescence", u.ID) |
| + } |
| + for d := dayOf(u.From); d <= dayOf(u.To); d++ { |
| + for _, o := range g.Owners { |
| + if !m.Holds[d][o] { |
| + return fmt.Errorf("reach: fleet %d's owner %s not held for U%d (fleet %d) day %d", g.ID, o, u.ID, p.ID, d) |
| + } |
| + } |
| + } |
| + } |
| + } |
| + } |
| + // liveness: a day with no uncertainty reaching the owner is complete |
| + owners := map[Owner]bool{} |
| + for _, f := range m.Fleets { |
| + for _, o := range f.Owners { |
| + owners[o] = true |
| + } |
| + } |
| + for o := range owners { |
| + for d := int64(0); d <= dayOf(m.Now); d++ { |
| + if m.Complete(o, d) { |
| + continue |
| + } |
| + if !m.reasonToHold(o, d) { |
| + return fmt.Errorf("liveness: %s day %d incomplete with no uncertainty reaching it", o, d) |
| + } |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +func (m *Model) reasonToHold(o Owner, d int64) bool { |
| + for _, g := range m.Fleets { |
| + mine := false |
| + for _, go_ := range g.Owners { |
| + if go_ == o { |
| + mine = true |
| + } |
| + } |
| + if !mine { |
| + continue |
| + } |
| + for p := g; p != nil; p = p.Pred { |
| + for _, u := range p.Uncertainties { |
| + if dayOf(u.From) <= d && d <= dayOf(u.To) { |
| + return true |
| + } |
| + } |
| + } |
| + } |
| + return false |
| +} |
amadan