From ddafe1b1aa75cb70d4b60a04f9fe0e806d94c18c Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 13:25:13 +0100 Subject: [PATCH 1/5] Build steps: a step records what it is, a restart ends the steps it left running, a tool ends one (issue 1890) --- go/buildstep/pull.go | 2 +- go/buildstep/stale.go | 166 +++++++++++++++++++++++++++++++++++++ go/buildstep/stale_test.go | 102 +++++++++++++++++++++++ go/buildstep/step.go | 6 +- go/delivery/steps.go | 3 + 5 files changed, 275 insertions(+), 4 deletions(-) create mode 100644 go/buildstep/stale.go create mode 100644 go/buildstep/stale_test.go diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index cc95414..80b41d8 100644 --- a/go/buildstep/pull.go +++ b/go/buildstep/pull.go @@ -314,7 +314,7 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) { _ = msg.Term() return } - step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID) + step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID, AskWhat(ask)) if err != nil { p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err) _ = msg.Nak() diff --git a/go/buildstep/stale.go b/go/buildstep/stale.go new file mode 100644 index 0000000..4f2324c --- /dev/null +++ b/go/buildstep/stale.go @@ -0,0 +1,166 @@ +package buildstep + +import ( + "encoding/json" + "fmt" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// **A step left running** (novox/hq issue 1890, note 5). A build module that stops while a step runs leaves its key in +// state running for good: nothing else writes that key's end, and the stream's redelivery only rewrites it when the same +// ask comes back. At its start a module ends the steps its node left running before the process began (EndStale), and +// its step-end tool ends one by hand (EndOne). + +// RestartReason is why a step a previous process left running is ended. +const RestartReason = "the build module restarted while it ran" + +// Scan reads a steps state: its keys and a key's stored value (nil for none). +type Scan struct { + Keys func() ([]string, error) + Get func(key string) (json.RawMessage, error) +} + +// AskWhat says what an ask is, in one line: its kind, and its repository, head and levels, or its module and +// fingerprint. +func AskWhat(a delivery.StepAsk) string { + short := func(s string) string { + if len(s) > 12 { + return s[:12] + } + return s + } + str := func(m map[string]any, k string) string { v, _ := m[k].(string); return v } + parts := []string{a.Kind + " ask"} + switch { + case a.Check != nil: + parts = append(parts, str(a.Check.Fetch, "repository")+"@"+short(str(a.Check.Fetch, "commit")), "levels "+strings.Join(a.Check.Levels, ",")) + if a.Check.Run != "" { + parts = append(parts, "run "+a.Check.Run) + } + case a.Build != nil: + parts = append(parts, "module "+a.Build.Module) + if r := str(a.Build.Fetch, "repository"); r != "" { + parts = append(parts, r+"@"+short(str(a.Build.Fetch, "commit"))) + } + if a.Build.FP != "" { + parts = append(parts, "fp "+short(a.Build.FP)) + } + case a.Compose != nil: + parts = append(parts, "compose", str(a.Compose.Fetch, "repository")+"@"+short(str(a.Compose.Fetch, "commit"))) + } + return strings.Join(parts, ", ") +} + +// endRunning writes step ended failed with why. +func endRunning(put Put, key string, step delivery.Step, why string) error { + step.State, step.Reason, step.Ended, step.Result = delivery.StepFailed, why, now().UTC(), nil + return write(put, key, step) +} + +// EndStale ends this node's steps that are running and started before `before` (the start of this process) as failed +// with RestartReason, and answers their ids. Only keys of node are read; a value that is no step is left alone. +func EndStale(scan Scan, put Put, node string, before time.Time, logf func(string, ...any)) ([]string, error) { + keys, err := scan.Keys() + if err != nil { + return nil, fmt.Errorf("buildstep: listing the steps of %s: %w", node, err) + } + var ended []string + for _, k := range keys { + n, _, perr := delivery.ParseStepKey(k) + if perr != nil || n != node { + continue + } + raw, err := scan.Get(k) + if err != nil { + return ended, fmt.Errorf("buildstep: reading the step %s: %w", k, err) + } + if len(raw) == 0 { + continue + } + step, err := delivery.DecodeStep(raw) + if err != nil || step.State != delivery.StepRunning || !step.Started.Before(before) { + continue + } + if err := endRunning(put, k, step, RestartReason); err != nil { + return ended, fmt.Errorf("buildstep: ending the step %s: %w", k, err) + } + if logf != nil { + logf("the step %s (%s) was left running by a previous process and is ended", k, step.Tool) + } + ended = append(ended, step.ID) + } + return ended, nil +} + +// EndOne ends this node's running step id (or the key .) as failed with why. A step that is not running is +// refused, and so is one live reports as held by this process. +func EndOne(scan Scan, put Put, node, id, why string, live func(id string) bool) (delivery.Step, error) { + id = strings.TrimSpace(id) + if n, rest, ok := strings.Cut(id, "."); ok { + if n != node { + return delivery.Step{}, fmt.Errorf("the step %q is of node %q, this module runs on %q", id, n, node) + } + id = rest + } + if strings.TrimSpace(why) == "" { + return delivery.Step{}, fmt.Errorf("say why the step is ended (why)") + } + key, err := delivery.StepKey(node, id) + if err != nil { + return delivery.Step{}, err + } + raw, err := scan.Get(key) + if err != nil { + return delivery.Step{}, err + } + if len(raw) == 0 { + return delivery.Step{}, fmt.Errorf("no step %s in the steps state", key) + } + step, err := delivery.DecodeStep(raw) + if err != nil { + return delivery.Step{}, err + } + if step.State != delivery.StepRunning { + return delivery.Step{}, fmt.Errorf("the step %s is %s, not running", key, step.State) + } + if live != nil && live(id) { + return delivery.Step{}, fmt.Errorf("the step %s is running in this process now; pause the seat or wait for it", key) + } + if err := endRunning(put, key, step, why); err != nil { + return delivery.Step{}, err + } + step.State, step.Reason = delivery.StepFailed, why + return step, nil +} + +// StepEndTool is the module's step-end tool, named name (build_go_step_end): it ends one step left running, with a +// reason, refusing one this process still runs. p may be nil. +func StepEndTool(name string, node string, scan Scan, put Put, p *Puller) stdio.Tool { + return stdio.Tool{Name: name, + Description: "End one step of this machine that is stuck in state running (for instance left by a restart of the module) " + + "as failed, with the reason given; its asker then sees it over. A step not running, or one this module runs now, is refused. " + + "id is the step id or the key ..", + Input: map[string]any{"type": "object", "required": []string{"id", "why"}, "properties": map[string]any{ + "id": map[string]any{"type": "string", "description": "the step id, or the key ."}, + "why": map[string]any{"type": "string", "description": "why it is ended; it becomes the step's reason"}}}, + Run: func(args map[string]any) (any, error) { + id, _ := args["id"].(string) + why, _ := args["why"].(string) + var live func(string) bool + if p != nil { + live = func(id string) bool { + for _, r := range p.Running() { + if r.ID == id { + return true + } + } + return false + } + } + return EndOne(scan, put, node, id, why, live) + }} +} diff --git a/go/buildstep/stale_test.go b/go/buildstep/stale_test.go new file mode 100644 index 0000000..e52f34c --- /dev/null +++ b/go/buildstep/stale_test.go @@ -0,0 +1,102 @@ +package buildstep + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// scanOf is the steps a keys double holds, last value per key. +func scanOf(k *keys) Scan { + return Scan{ + Keys: func() ([]string, error) { + k.mu.Lock() + defer k.mu.Unlock() + var out []string + for key := range k.puts { + out = append(out, key) + } + return out, nil + }, + Get: func(key string) (json.RawMessage, error) { + k.mu.Lock() + defer k.mu.Unlock() + s, ok := k.puts[key] + if !ok { + return nil, nil + } + raw, err := delivery.EncodeStep(s[len(s)-1]) + return raw, err + }, + } +} + +// 1890: EndStale ends only this node's steps left running before the process began; EndOne ends one by hand. +func TestEndStaleAndEndOne(t *testing.T) { + k := newKeys() + boot := time.Now() + put := func(node, id, state string, started time.Time) { + key, _ := delivery.StepKey(node, id) + _, _ = k.put(key, json.RawMessage(mustStep(t, delivery.Step{ID: id, Tool: "check ask", Node: node, State: state, Started: started}))) + } + put("ace", "old1", delivery.StepRunning, boot.Add(-time.Hour)) + put("ace", "new1", delivery.StepRunning, boot.Add(time.Second)) + put("ace", "done1", delivery.StepPassed, boot.Add(-time.Hour)) + put("g14", "other1", delivery.StepRunning, boot.Add(-time.Hour)) + scan := scanOf(k) + ended, err := EndStale(scan, k.put, "ace", boot, nil) + if err != nil || len(ended) != 1 || ended[0] != "old1" { + t.Fatalf("ended %v, %v", ended, err) + } + if s := k.last("ace.old1"); s.State != delivery.StepFailed || s.Reason != RestartReason || s.Ended.IsZero() { + t.Errorf("old1 = %+v", s) + } + for key, state := range map[string]string{"ace.new1": delivery.StepRunning, "ace.done1": delivery.StepPassed, "g14.other1": delivery.StepRunning} { + if s := k.last(key); s.State != state { + t.Errorf("%s = %s, want %s", key, s.State, state) + } + } + // EndOne: refuses without why, a held step, a step not running and another node's; ends the rest. + if _, err := EndOne(scan, k.put, "ace", "new1", "", nil); err == nil { + t.Error("ended with no why") + } + if _, err := EndOne(scan, k.put, "ace", "new1", "stuck", func(string) bool { return true }); err == nil || !strings.Contains(err.Error(), "running in this process") { + t.Errorf("held step: %v", err) + } + if _, err := EndOne(scan, k.put, "ace", "done1", "x", nil); err == nil { + t.Error("ended a passed step") + } + if _, err := EndOne(scan, k.put, "ace", "g14.other1", "x", nil); err == nil { + t.Error("ended another node's step") + } + if _, err := EndOne(scan, k.put, "ace", "ace.new1", "stuck for good", nil); err != nil { + t.Fatal(err) + } + if s := k.last("ace.new1"); s.State != delivery.StepFailed || s.Reason != "stuck for good" { + t.Errorf("new1 = %+v", s) + } +} + +func mustStep(t *testing.T, s delivery.Step) []byte { + t.Helper() + raw, err := delivery.EncodeStep(s) + if err != nil { + t.Fatal(err) + } + return raw +} + +// 1890: AskWhat names the repository, head and levels of a check ask and the module of a build ask. +func TestAskWhat(t *testing.T) { + check := delivery.StepAsk{Kind: "check", Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/mesh-sdk", "commit": "0123456789abcdef0123"}, Levels: []string{"unit", "integration"}}} + if got, want := AskWhat(check), "check ask, novox/mesh-sdk@0123456789ab, levels unit,integration"; got != want { + t.Errorf("check = %q", got) + } + build := delivery.StepAsk{Kind: "delivery", Build: &delivery.BuildJob{Module: "mesh-x", FP: "abcdef0123456789"}} + if got, want := AskWhat(build), "delivery ask, module mesh-x, fp abcdef012345"; got != want { + t.Errorf("build = %q", got) + } +} diff --git a/go/buildstep/step.go b/go/buildstep/step.go index 8a01aaa..0e26fe1 100644 --- a/go/buildstep/step.go +++ b/go/buildstep/step.go @@ -43,7 +43,7 @@ func NewID() string { // the error's text, or refused when the error is a *Refusal; a panic in run ends the step failed, naming it, and // the process goes on. If the running write fails, nothing was started and Start answers the error. func Start(ctx context.Context, put Put, node, tool string, run func(ctx context.Context) (result any, err error)) (string, error) { - step, key, err := begin(put, node, tool, NewID()) + step, key, err := begin(put, node, tool, NewID(), "") if err != nil { return "", err } @@ -52,8 +52,8 @@ func Start(ctx context.Context, put Put, node, tool string, run func(ctx context } // begin writes the step running under id. -func begin(put Put, node, tool, id string) (delivery.Step, string, error) { - step := delivery.Step{ID: id, Tool: tool, Node: node, State: delivery.StepRunning, Started: now().UTC()} +func begin(put Put, node, tool, id, what string) (delivery.Step, string, error) { + step := delivery.Step{ID: id, Tool: tool, Node: node, State: delivery.StepRunning, Started: now().UTC(), What: what} key, err := delivery.StepKey(node, step.ID) if err != nil { return delivery.Step{}, "", err diff --git a/go/delivery/steps.go b/go/delivery/steps.go index 61f8114..917278a 100644 --- a/go/delivery/steps.go +++ b/go/delivery/steps.go @@ -70,6 +70,9 @@ type Step struct { Ended time.Time `json:"ended,omitzero"` Reason string `json:"reason,omitempty"` Result json.RawMessage `json:"result,omitempty"` + // What says what the step is, for a reader of the steps state: an ask's kind, repository, head and levels or + // module (issue 1890). Empty for a step a tool started by hand. + What string `json:"what,omitempty"` } func (s Step) check() error { From 18e0a01c54885de4807384a52c57017ad0c507fb Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 13:25:58 +0100 Subject: [PATCH 2/5] Build steps: EndStaleAtStart waits for the runtime and ends the left-running steps (issue 1890) --- go/buildstep/stale.go | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/go/buildstep/stale.go b/go/buildstep/stale.go index 4f2324c..92a33f7 100644 --- a/go/buildstep/stale.go +++ b/go/buildstep/stale.go @@ -1,6 +1,7 @@ package buildstep import ( + "context" "encoding/json" "fmt" "strings" @@ -164,3 +165,40 @@ func StepEndTool(name string, node string, scan Scan, put Put, p *Puller) stdio. return EndOne(scan, put, node, id, why, live) }} } + +// StaleAttempts and StaleRetry bound EndStaleAtStart's tries: the steps state is readable once the runtime serves. +var ( + StaleAttempts = 6 + StaleRetry = 5 * time.Second +) + +// EndStaleAtStart is what a build module runs once at its start, in a goroutine: it waits for the runtime to serve, +// then ends this node's steps left running before `before` (EndStale), trying again when the state cannot be read yet. +// A step this process starts is never one of them: it starts after `before`. +func EndStaleAtStart(ctx context.Context, scan Scan, put Put, node string, before time.Time, logf func(string, ...any)) { + select { + case <-ctx.Done(): + return + case <-stdio.Served(): + } + for attempt := 1; ; attempt++ { + ended, err := EndStale(scan, put, node, before, logf) + if err == nil { + if logf != nil { + logf("%d steps of %s left running by a previous process ended", len(ended), node) + } + return + } + if logf != nil { + logf("ending the steps left running (try %d of %d): %v", attempt, StaleAttempts, err) + } + if attempt >= StaleAttempts { + return + } + select { + case <-ctx.Done(): + return + case <-time.After(StaleRetry): + } + } +} From 6b1c7a5613b120e6dc70db7e3c93a601a3ac0643 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 13:42:50 +0100 Subject: [PATCH 3/5] Build steps: a restart leaves pulled asks to the stream, ends steps by compare-and-set, and step_end refuses a step this process started (issue 1890 review 1) --- go/buildstep/stale.go | 85 ++++++++++++++++++++++++++++---------- go/buildstep/stale_test.go | 84 +++++++++++++++++++++++-------------- 2 files changed, 117 insertions(+), 52 deletions(-) diff --git a/go/buildstep/stale.go b/go/buildstep/stale.go index 92a33f7..0615346 100644 --- a/go/buildstep/stale.go +++ b/go/buildstep/stale.go @@ -3,6 +3,7 @@ package buildstep import ( "context" "encoding/json" + "errors" "fmt" "strings" "time" @@ -19,10 +20,36 @@ import ( // RestartReason is why a step a previous process left running is ended. const RestartReason = "the build module restarted while it ran" -// Scan reads a steps state: its keys and a key's stored value (nil for none). +// Boot is when this process began. A step started at or after it belongs to this process, which may still run it, so +// EndOne refuses it. +var Boot = time.Now() + +// Scan reads and writes a steps state with its revisions: its keys, a key's stored value and revision (nil and 0 for +// none), and a write that lands only while the key is still at the revision read (KeptState.Update), failing with +// stdio.ErrChanged when another writer was first. type Scan struct { - Keys func() ([]string, error) - Get func(key string) (json.RawMessage, error) + Keys func() ([]string, error) + Get func(key string) (json.RawMessage, uint64, error) + Update func(key string, value any, revision uint64) (uint64, error) +} + +// ScanOf is the Scan of a state the process keeps. +func ScanOf(s stdio.KeptState) Scan { + return Scan{Keys: s.Keys, Update: s.Update, Get: func(key string) (json.RawMessage, uint64, error) { + e, err := s.Get(key) + if err != nil || e == nil { + return nil, 0, err + } + return e.Value, e.Revision, nil + }} +} + +// isAsk says a step is a pulled ask's: its key is . and its tool is " ask" (a hand-started step +// has the tool's name). An ask's key is the stream's to settle, never a restart's: the puller leaves an ask it was +// running unacknowledged so the stream redelivers it, and the asker's wait takes the first end it reads on the key. +func isAsk(key string, s delivery.Step) bool { + _, _, err := delivery.ParseAskStepKey(key) + return err == nil && strings.HasSuffix(s.Tool, " ask") } // AskWhat says what an ask is, in one line: its kind, and its repository, head and levels, or its module and @@ -56,15 +83,22 @@ func AskWhat(a delivery.StepAsk) string { return strings.Join(parts, ", ") } -// endRunning writes step ended failed with why. -func endRunning(put Put, key string, step delivery.Step, why string) error { +// endRunning writes step ended failed with why, only while its key is still at revision. +func endRunning(scan Scan, key string, step delivery.Step, revision uint64, why string) error { step.State, step.Reason, step.Ended, step.Result = delivery.StepFailed, why, now().UTC(), nil - return write(put, key, step) + raw, err := delivery.EncodeStep(step) + if err != nil { + return err + } + _, err = scan.Update(key, json.RawMessage(raw), revision) + return err } // EndStale ends this node's steps that are running and started before `before` (the start of this process) as failed -// with RestartReason, and answers their ids. Only keys of node are read; a value that is no step is left alone. -func EndStale(scan Scan, put Put, node string, before time.Time, logf func(string, ...any)) ([]string, error) { +// with RestartReason, and answers their ids. Only keys of node are read; a value that is no step is left alone. A step +// of a pulled ask is left to the stream's redelivery (isAsk), and a key written by another writer since it was read +// (a redelivered ask's start, a real end) is skipped. +func EndStale(scan Scan, node string, before time.Time, logf func(string, ...any)) ([]string, error) { keys, err := scan.Keys() if err != nil { return nil, fmt.Errorf("buildstep: listing the steps of %s: %w", node, err) @@ -75,7 +109,7 @@ func EndStale(scan Scan, put Put, node string, before time.Time, logf func(strin if perr != nil || n != node { continue } - raw, err := scan.Get(k) + raw, rev, err := scan.Get(k) if err != nil { return ended, fmt.Errorf("buildstep: reading the step %s: %w", k, err) } @@ -83,10 +117,13 @@ func EndStale(scan Scan, put Put, node string, before time.Time, logf func(strin continue } step, err := delivery.DecodeStep(raw) - if err != nil || step.State != delivery.StepRunning || !step.Started.Before(before) { + if err != nil || step.State != delivery.StepRunning || !step.Started.Before(before) || isAsk(k, step) { continue } - if err := endRunning(put, k, step, RestartReason); err != nil { + if err := endRunning(scan, k, step, rev, RestartReason); err != nil { + if errors.Is(err, stdio.ErrChanged) { + continue + } return ended, fmt.Errorf("buildstep: ending the step %s: %w", k, err) } if logf != nil { @@ -98,8 +135,9 @@ func EndStale(scan Scan, put Put, node string, before time.Time, logf func(strin } // EndOne ends this node's running step id (or the key .) as failed with why. A step that is not running is -// refused, and so is one live reports as held by this process. -func EndOne(scan Scan, put Put, node, id, why string, live func(id string) bool) (delivery.Step, error) { +// refused, and so is one started since this process began (Boot: it may still run, whether a pulled ask or a step a +// tool started), one live reports as held, and one whose key changed since it was read. +func EndOne(scan Scan, node, id, why string, live func(id string) bool) (delivery.Step, error) { id = strings.TrimSpace(id) if n, rest, ok := strings.Cut(id, "."); ok { if n != node { @@ -114,7 +152,7 @@ func EndOne(scan Scan, put Put, node, id, why string, live func(id string) bool) if err != nil { return delivery.Step{}, err } - raw, err := scan.Get(key) + raw, rev, err := scan.Get(key) if err != nil { return delivery.Step{}, err } @@ -128,10 +166,13 @@ func EndOne(scan Scan, put Put, node, id, why string, live func(id string) bool) if step.State != delivery.StepRunning { return delivery.Step{}, fmt.Errorf("the step %s is %s, not running", key, step.State) } - if live != nil && live(id) { - return delivery.Step{}, fmt.Errorf("the step %s is running in this process now; pause the seat or wait for it", key) + if !step.Started.Before(Boot) || (live != nil && live(id)) { + return delivery.Step{}, fmt.Errorf("the step %s was started by this process and may run now; only a step a previous process left is ended", key) } - if err := endRunning(put, key, step, why); err != nil { + if err := endRunning(scan, key, step, rev, why); err != nil { + if errors.Is(err, stdio.ErrChanged) { + return delivery.Step{}, fmt.Errorf("the step %s changed while it was being ended; read it again", key) + } return delivery.Step{}, err } step.State, step.Reason = delivery.StepFailed, why @@ -139,8 +180,8 @@ func EndOne(scan Scan, put Put, node, id, why string, live func(id string) bool) } // StepEndTool is the module's step-end tool, named name (build_go_step_end): it ends one step left running, with a -// reason, refusing one this process still runs. p may be nil. -func StepEndTool(name string, node string, scan Scan, put Put, p *Puller) stdio.Tool { +// reason, refusing one this process started. p may be nil. +func StepEndTool(name string, node string, scan Scan, p *Puller) stdio.Tool { return stdio.Tool{Name: name, Description: "End one step of this machine that is stuck in state running (for instance left by a restart of the module) " + "as failed, with the reason given; its asker then sees it over. A step not running, or one this module runs now, is refused. " + @@ -162,7 +203,7 @@ func StepEndTool(name string, node string, scan Scan, put Put, p *Puller) stdio. return false } } - return EndOne(scan, put, node, id, why, live) + return EndOne(scan, node, id, why, live) }} } @@ -175,14 +216,14 @@ var ( // EndStaleAtStart is what a build module runs once at its start, in a goroutine: it waits for the runtime to serve, // then ends this node's steps left running before `before` (EndStale), trying again when the state cannot be read yet. // A step this process starts is never one of them: it starts after `before`. -func EndStaleAtStart(ctx context.Context, scan Scan, put Put, node string, before time.Time, logf func(string, ...any)) { +func EndStaleAtStart(ctx context.Context, scan Scan, node string, before time.Time, logf func(string, ...any)) { select { case <-ctx.Done(): return case <-stdio.Served(): } for attempt := 1; ; attempt++ { - ended, err := EndStale(scan, put, node, before, logf) + ended, err := EndStale(scan, node, before, logf) if err == nil { if logf != nil { logf("%d steps of %s left running by a previous process ended", len(ended), node) diff --git a/go/buildstep/stale_test.go b/go/buildstep/stale_test.go index e52f34c..e322e54 100644 --- a/go/buildstep/stale_test.go +++ b/go/buildstep/stale_test.go @@ -2,15 +2,16 @@ package buildstep import ( "encoding/json" - "strings" + stdio "git.novox.be/novox/mesh-sdk/go" "testing" "time" "git.novox.be/novox/mesh-sdk/go/delivery" ) -// scanOf is the steps a keys double holds, last value per key. -func scanOf(k *keys) Scan { +// scanOf is the steps a keys double holds, last value per key, its revision the count of its puts; a write through +// Update lands only at the revision read, as the state's does. race, when set, runs once before the next Update's check. +func scanOf(k *keys, race func(key string)) Scan { return Scan{ Keys: func() ([]string, error) { k.mu.Lock() @@ -21,62 +22,85 @@ func scanOf(k *keys) Scan { } return out, nil }, - Get: func(key string) (json.RawMessage, error) { + Get: func(key string) (json.RawMessage, uint64, error) { k.mu.Lock() defer k.mu.Unlock() s, ok := k.puts[key] if !ok { - return nil, nil + return nil, 0, nil } raw, err := delivery.EncodeStep(s[len(s)-1]) - return raw, err + return raw, uint64(len(s)), err + }, + Update: func(key string, value any, revision uint64) (uint64, error) { + if race != nil { + race(key) + } + k.mu.Lock() + n := uint64(len(k.puts[key])) + k.mu.Unlock() + if n != revision { + return 0, stdio.ErrChanged + } + return k.put(key, value) }, } } -// 1890: EndStale ends only this node's steps left running before the process began; EndOne ends one by hand. +// 1890: EndStale ends only this node's hand-started steps left running before the process began, never a pulled ask's +// (the stream redelivers it) and never a key written meanwhile; EndOne ends one by hand, never one this process started. func TestEndStaleAndEndOne(t *testing.T) { k := newKeys() boot := time.Now() - put := func(node, id, state string, started time.Time) { + oldBoot := Boot + Boot = boot + defer func() { Boot = oldBoot }() + put := func(node, id, tool, state string, started time.Time) { key, _ := delivery.StepKey(node, id) - _, _ = k.put(key, json.RawMessage(mustStep(t, delivery.Step{ID: id, Tool: "check ask", Node: node, State: state, Started: started}))) + _, _ = k.put(key, json.RawMessage(mustStep(t, delivery.Step{ID: id, Tool: tool, Node: node, State: state, Started: started}))) } - put("ace", "old1", delivery.StepRunning, boot.Add(-time.Hour)) - put("ace", "new1", delivery.StepRunning, boot.Add(time.Second)) - put("ace", "done1", delivery.StepPassed, boot.Add(-time.Hour)) - put("g14", "other1", delivery.StepRunning, boot.Add(-time.Hour)) - scan := scanOf(k) - ended, err := EndStale(scan, k.put, "ace", boot, nil) + put("ace", "old1", "build_go_fetch", delivery.StepRunning, boot.Add(-time.Hour)) + put("ace", "askold", "check ask", delivery.StepRunning, boot.Add(-time.Hour)) + put("ace", "raced1", "build_go_test", delivery.StepRunning, boot.Add(-time.Hour)) + put("ace", "new1", "build_go_fetch", delivery.StepRunning, boot.Add(time.Second)) + put("ace", "done1", "build_go_fetch", delivery.StepPassed, boot.Add(-time.Hour)) + put("g14", "other1", "build_go_fetch", delivery.StepRunning, boot.Add(-time.Hour)) + // raced1 is ended by its own run between the read and the write. + raced := false + scan := scanOf(k, func(key string) { + if key == "ace.raced1" && !raced { + raced = true + put("ace", "raced1", "build_go_test", delivery.StepPassed, boot.Add(-time.Hour)) + } + }) + ended, err := EndStale(scan, "ace", boot, nil) if err != nil || len(ended) != 1 || ended[0] != "old1" { t.Fatalf("ended %v, %v", ended, err) } if s := k.last("ace.old1"); s.State != delivery.StepFailed || s.Reason != RestartReason || s.Ended.IsZero() { t.Errorf("old1 = %+v", s) } - for key, state := range map[string]string{"ace.new1": delivery.StepRunning, "ace.done1": delivery.StepPassed, "g14.other1": delivery.StepRunning} { + for key, state := range map[string]string{"ace.askold": delivery.StepRunning, "ace.raced1": delivery.StepPassed, "ace.new1": delivery.StepRunning, "ace.done1": delivery.StepPassed, "g14.other1": delivery.StepRunning} { if s := k.last(key); s.State != state { t.Errorf("%s = %s, want %s", key, s.State, state) } } - // EndOne: refuses without why, a held step, a step not running and another node's; ends the rest. - if _, err := EndOne(scan, k.put, "ace", "new1", "", nil); err == nil { - t.Error("ended with no why") + // EndOne: refuses without why, a held step, one started by this process (a pulled ask or a step tool's), a step not + // running and another node's; ends a step a previous process left. + for _, c := range []struct{ id, why string }{{"old1", ""}, {"new1", "stuck"}, {"done1", "x"}, {"g14.other1", "x"}} { + if _, err := EndOne(scan, "ace", c.id, c.why, nil); err == nil { + t.Errorf("ended %s with why %q", c.id, c.why) + } } - if _, err := EndOne(scan, k.put, "ace", "new1", "stuck", func(string) bool { return true }); err == nil || !strings.Contains(err.Error(), "running in this process") { - t.Errorf("held step: %v", err) + put("ace", "old2", "build_go_fetch", delivery.StepRunning, boot.Add(-time.Hour)) + if _, err := EndOne(scan, "ace", "old2", "x", func(string) bool { return true }); err == nil { + t.Error("ended a held step") } - if _, err := EndOne(scan, k.put, "ace", "done1", "x", nil); err == nil { - t.Error("ended a passed step") - } - if _, err := EndOne(scan, k.put, "ace", "g14.other1", "x", nil); err == nil { - t.Error("ended another node's step") - } - if _, err := EndOne(scan, k.put, "ace", "ace.new1", "stuck for good", nil); err != nil { + if _, err := EndOne(scan, "ace", "ace.old2", "stuck for good", nil); err != nil { t.Fatal(err) } - if s := k.last("ace.new1"); s.State != delivery.StepFailed || s.Reason != "stuck for good" { - t.Errorf("new1 = %+v", s) + if s := k.last("ace.old2"); s.State != delivery.StepFailed || s.Reason != "stuck for good" { + t.Errorf("old2 = %+v", s) } } From e66c6fea6204528cc394dfd0bb008889b4eec5df Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 13:43:05 +0100 Subject: [PATCH 4/5] Build steps: test the ask key with a real ask id (issue 1890) --- go/buildstep/stale_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/go/buildstep/stale_test.go b/go/buildstep/stale_test.go index e322e54..7d81c8d 100644 --- a/go/buildstep/stale_test.go +++ b/go/buildstep/stale_test.go @@ -60,7 +60,7 @@ func TestEndStaleAndEndOne(t *testing.T) { _, _ = k.put(key, json.RawMessage(mustStep(t, delivery.Step{ID: id, Tool: tool, Node: node, State: state, Started: started}))) } put("ace", "old1", "build_go_fetch", delivery.StepRunning, boot.Add(-time.Hour)) - put("ace", "askold", "check ask", delivery.StepRunning, boot.Add(-time.Hour)) + put("ace", "mjk3l9x0abcdef", "check ask", delivery.StepRunning, boot.Add(-time.Hour)) put("ace", "raced1", "build_go_test", delivery.StepRunning, boot.Add(-time.Hour)) put("ace", "new1", "build_go_fetch", delivery.StepRunning, boot.Add(time.Second)) put("ace", "done1", "build_go_fetch", delivery.StepPassed, boot.Add(-time.Hour)) @@ -80,7 +80,7 @@ func TestEndStaleAndEndOne(t *testing.T) { if s := k.last("ace.old1"); s.State != delivery.StepFailed || s.Reason != RestartReason || s.Ended.IsZero() { t.Errorf("old1 = %+v", s) } - for key, state := range map[string]string{"ace.askold": delivery.StepRunning, "ace.raced1": delivery.StepPassed, "ace.new1": delivery.StepRunning, "ace.done1": delivery.StepPassed, "g14.other1": delivery.StepRunning} { + for key, state := range map[string]string{"ace.mjk3l9x0abcdef": delivery.StepRunning, "ace.raced1": delivery.StepPassed, "ace.new1": delivery.StepRunning, "ace.done1": delivery.StepPassed, "g14.other1": delivery.StepRunning} { if s := k.last(key); s.State != state { t.Errorf("%s = %s, want %s", key, s.State, state) } From 6b8497314297b83d8db8249591b5bf8b32ecb930 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 13:51:54 +0100 Subject: [PATCH 5/5] Build steps: an ask key past the step bound is ended, at start and on the module's round (issue 1890 review 2) --- go/buildstep/stale.go | 37 ++++++++++++++++++++++++++++++++----- go/buildstep/stale_test.go | 13 +++++++++---- go/delivery/stepasks.go | 5 +++++ 3 files changed, 46 insertions(+), 9 deletions(-) diff --git a/go/buildstep/stale.go b/go/buildstep/stale.go index 0615346..4147934 100644 --- a/go/buildstep/stale.go +++ b/go/buildstep/stale.go @@ -47,6 +47,7 @@ func ScanOf(s stdio.KeptState) Scan { // isAsk says a step is a pulled ask's: its key is . and its tool is " ask" (a hand-started step // has the tool's name). An ask's key is the stream's to settle, never a restart's: the puller leaves an ask it was // running unacknowledged so the stream redelivers it, and the asker's wait takes the first end it reads on the key. +// Only once the asker's bound has passed (delivery.StepBound) is the key nobody's, and then it is ended. func isAsk(key string, s delivery.Step) bool { _, _, err := delivery.ParseAskStepKey(key) return err == nil && strings.HasSuffix(s.Tool, " ask") @@ -96,7 +97,9 @@ func endRunning(scan Scan, key string, step delivery.Step, revision uint64, why // EndStale ends this node's steps that are running and started before `before` (the start of this process) as failed // with RestartReason, and answers their ids. Only keys of node are read; a value that is no step is left alone. A step -// of a pulled ask is left to the stream's redelivery (isAsk), and a key written by another writer since it was read +// of a pulled ask is left to the stream's redelivery (isAsk) until its start is older than delivery.StepBound: its asker +// stopped waiting then, whether the stream gave it to another node or gave it up, and it is ended as failed, "past the +// step bound: its asker stopped waiting at