ENG-9: the node seat's read verb pull-waits answers which failed applies wait on which store's holder and the last retry with its trigger (design 51, issue 1966; needs the verb in the node-engine seat's definition)

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
2026-11-04 12:15:26 +01:00
parent 7f33c8b499
commit c93feeb30f
6 changed files with 63 additions and 0 deletions
+11
View File
@@ -284,6 +284,17 @@ const compositionReadFor = 20 * time.Second
// compositionFrom answers the verb composition from the node agent that runs now (issue 774); refused in words while
// none does.
// pullWaitsFrom answers the verb pull-waits from the agent running now (ENG-9).
func pullWaitsFrom(current *agentNow) func() (any, error) {
return func() (any, error) {
ag := current.get()
if ag == nil {
return nil, errors.New("this engine's node agent has not started, so it waits on no store yet")
}
return map[string]any{"waits": ag.PullWaits()}, nil
}
}
func compositionFrom(current *agentNow) func(module, digest string, diff bool) (any, error) {
return func(module, digest string, diff bool) (any, error) {
ag := current.get()
+1
View File
@@ -1015,6 +1015,7 @@ func runLink(ctx context.Context, opts options) error {
go reconcileActs(ctx, queue.Acts)
current := &agentNow{}
queue.Acts.Composition = compositionFrom(current)
queue.Acts.PullWaits = pullWaitsFrom(current)
queue.Acts.Compose = composeFrom(current, &composerNow)
queue.Acts.Packages = packagesFrom(mine.Node, queue, &composerNow)
queue.Acts.ComposeCandidate = composeCandidateFrom(mine.Node, &shadowNow)
+28
View File
@@ -1932,3 +1932,31 @@ func (a *Agent) Settled(holder string) {
a.askPullAgain(module, asks[i])
}
}
// PullWaitView is one failed apply that waits on the holder of a store, as the engine's read verb pull-waits shows it
// (ENG-9, design 51): the module, the image and the seat it came through, whose settling it waits on, when the pull began,
// how many times that holder has settled since this engine started and when last, and what triggered the last retry.
type PullWaitView struct {
Module string `json:"module"`
Image string `json:"image"`
Seat string `json:"seat,omitempty"`
Holder string `json:"holder"`
Started time.Time `json:"pull-started"`
Settled int `json:"holder-settled"`
SettledAt time.Time `json:"holder-last-settled,omitzero"`
LastRetry string `json:"last-retry,omitempty"`
LastRetryAt time.Time `json:"last-retry-at,omitzero"`
}
// PullWaits are the failed applies that wait on a store's holder, by module.
func (a *Agent) PullWaits() []PullWaitView {
a.mu.Lock()
defer a.mu.Unlock()
out := []PullWaitView{}
for module, pw := range a.pulls {
out = append(out, PullWaitView{Module: module, Image: pw.Image, Seat: pw.Seat, Holder: pw.Holder, Started: pw.Started,
Settled: a.settledSeq[pw.Holder], SettledAt: a.settledAt[pw.Holder], LastRetry: pw.last, LastRetryAt: pw.lastAt})
}
sort.Slice(out, func(i, j int) bool { return out[i].Module < out[j].Module })
return out
}
+8
View File
@@ -521,6 +521,11 @@ func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) {
if n := len(b.appliedList()); n != 2 {
t.Fatalf("a second failure after the same settled started a further apply (%d applies)", n)
}
// The read verb's view: whom it waits on, how often that holder settled, and what triggered the last retry.
if w := b.ag.PullWaits(); len(w) != 1 || w[0].Module != "web" || w[0].Holder != "distribution" || w[0].Image != "reg/x@sha256:1" ||
w[0].Settled != 1 || !strings.Contains(w[0].LastRetry, "distribution settled") {
t.Fatalf("the engine says it waits on %+v", w)
}
// The next settled answers once more, and the pull works: the module runs, and waits on nothing.
b.mu.Lock()
b.failWith = ""
@@ -529,6 +534,9 @@ func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) {
b.ag.Settled("distribution")
eventually(t, "not tried again on the next settled", func() bool { return len(b.appliedList()) == 3 })
eventually(t, "not running", func() bool { return b.haves.last("web").Phase == string(delivery.PhaseRunning) })
if w := b.ag.PullWaits(); len(w) != 0 {
t.Fatalf("a running module still waits on %+v", w)
}
b.ag.Settled("distribution")
time.Sleep(50 * time.Millisecond)
if n := len(b.appliedList()); n != 3 {
+10
View File
@@ -47,6 +47,9 @@ const (
ActPackages = "packages"
// ActComposeCandidate composes a check's candidate on this node in shadow, read-only (novox/hq ADR 0593 §7).
ActComposeCandidate = "compose-candidate"
// ActPullWaits reads the failed applies that wait on a store's holder, and the last retry with its trigger (design 51,
// issue 1966 ENG-9).
ActPullWaits = "pull-waits"
)
// The states of an act.
@@ -149,6 +152,8 @@ type HandActs struct {
// module, applying and writing nothing; with input, also the composition input it composed from (the verb
// compose-candidate, ADR 0593 §7).
ComposeCandidate func(candidate string, input bool) (any, error)
// PullWaits answers which failed applies wait on which store's holder, and the last retry of each with what triggered it.
PullWaits func() (any, error)
// Say says one of the seat's events, asked, done or refused; nil says nothing.
Say func(event string, body any)
Now func() time.Time
@@ -230,6 +235,11 @@ func (h *HandActs) Call(c EngineCall) (any, error) {
input = b
}
return h.ComposeCandidate(arg("candidate"), input)
case ActPullWaits:
if h.PullWaits == nil {
return nil, errors.New("this engine waits on no store yet: its node agent has not started")
}
return h.PullWaits()
case "acts":
if id := arg("id"); id != "" {
return h.Act(id)
+5
View File
@@ -223,6 +223,10 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
return map[string]any{"module": module, "digest": digest, "diff": diff}, nil
},
Compose: func(module string) (any, error) { return map[string]any{"composed": true, "retried": module}, nil },
// ENG-9: the failed applies that wait on a store's holder.
PullWaits: func() (any, error) {
return map[string]any{"waits": []map[string]any{{"module": "web", "holder": "distribution", "last-retry": "distribution settled"}}}, nil
},
// ADR 0565 §6, §15: the node's key of the engine's state packages.
Packages: func(write bool) (any, error) {
key := []byte(`{"node":"n1","packages":[{"package":"curl","modules":["backup","web"],"absent":false}]}`)
@@ -245,6 +249,7 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
// The recovery verb compose (issue 774) makes a pass now, with a module's apply tried again when named.
{"compose", EngineToolSubject("compose", "n1"), `"composed":true,"retried":""`},
{"compose", EngineToolSubject("compose", "n1") + " " + `{"module":"route-proxy"}`, `"retried":"route-proxy"`},
{"pull-waits", EngineToolSubject("pull-waits", "n1"), `"holder":"distribution","last-retry":"distribution settled","module":"web"`},
// The read verb packages (ADR 0565 §6) names its source; with write, what it wrote, and a key the engine has not
// written is said and not an error.
{"packages", EngineToolSubject("packages", "n1"), `"source":"mesh-host.packages","key":{"node":"n1"`},