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) } }