novox/mesh-sdk
18
0
Fork 0
Code Issues 1 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

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)

Browse Source
This commit is contained in:
jschoubben
2026-10-26 13:42:50 +01:00
parent 18e0a01c54
commit 6b1c7a5613
2 changed files with 117 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/buildstep/stale.go
+63 -22
View File
@@ -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 <node>.<ask id> and its tool is "<kind> 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 <node>.<id>) 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)
go/buildstep/stale_test.go
+54 -30
View File
@@ -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)
}
}