Merge main (issue 1890's step records) into feat/890-placement: a decline begins its step with what it is

go/buildstep/pull.go

@@ -319,7 +319,7 @@ func (p *Puller) next(ctx context.Context, wait time.Duration) (*Pulled, error)
 // message: the asker puts the ask again under a new id (issue 890). A message whose decline cannot be written is handed
 // back, as one whose start cannot be.
 func (p *Puller) decline(msg *Pulled, ask delivery.StepAsk, lacks []string) {
-	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 {
 		why := Refuse("the ask %s needs %s, and %s lacks %s: it is left to a node that has it", ask.ID,
 			strings.Join(ask.Needs, ", "), p.Node, strings.Join(lacks, ", "))
@@ -369,7 +369,7 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
 		p.decline(msg, ask, lacks)
 		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()

go/buildstep/stale.go

@@ -0,0 +1,272 @@
+package buildstep
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"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"
+
+// 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, 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.
+// 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")
+}
+
+// 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, 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
+	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. A step
+// 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 <time>". 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)
+	}
+	var ended []string
+	for _, k := range keys {
+		n, _, perr := delivery.ParseStepKey(k)
+		if perr != nil || n != node {
+			continue
+		}
+		raw, rev, 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 {
+			continue
+		}
+		why := RestartReason
+		if isAsk(k, step) {
+			if !step.Started.Add(delivery.StepBound).Before(now()) {
+				continue
+			}
+			why = "past the step bound: its asker stopped waiting at " + step.Started.Add(delivery.StepBound).UTC().Format("2006-01-02T15:04:05Z")
+		} else if !step.Started.Before(before) {
+			continue
+		}
+		if err := endRunning(scan, k, step, rev, why); err != nil {
+			if errors.Is(err, stdio.ErrChanged) {
+				continue
+			}
+			return ended, fmt.Errorf("buildstep: ending the step %s: %w", k, err)
+		}
+		if logf != nil {
+			logf("the step %s (%s) was left running and is ended: %s", k, step.Tool, why)
+		}
+		ended = append(ended, step.ID)
+	}
+	return ended, nil
+}
+
+// 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 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 {
+			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, rev, 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 !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(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
+	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 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, so ending an ask still in its stream fails its asker. " +
+			"A step not running, or one this process started, is refused. " +
+			"id is the step id or the key <node>.<id>.",
+		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 <node>.<id>"},
+			"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, 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, 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, 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):
+		}
+	}
+}
+
+// EndStaleRound is EndStale as the module's periodic round runs it: at once, and only when the runtime serves (a round
+// before that is skipped, the start's EndStaleAtStart and the next round cover it). It ends the steps a previous
+// process left and the asks past delivery.StepBound.
+func EndStaleRound(scan Scan, node string, logf func(string, ...any)) {
+	select {
+	case <-stdio.Served():
+	default:
+		return
+	}
+	if _, err := EndStale(scan, node, Boot, logf); err != nil && logf != nil {
+		logf("ending the steps left running: %v", err)
+	}
+}

go/buildstep/stale_test.go

@@ -0,0 +1,131 @@
+package buildstep
+
+import (
+	"encoding/json"
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// 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()
+			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, uint64, error) {
+			k.mu.Lock()
+			defer k.mu.Unlock()
+			s, ok := k.puts[key]
+			if !ok {
+				return nil, 0, nil
+			}
+			raw, err := delivery.EncodeStep(s[len(s)-1])
+			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 hand-started steps left running before the process began and the asks past the
+// step bound (a younger ask is the stream's to redeliver), 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()
+	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: tool, Node: node, State: state, Started: started})))
+	}
+	put("ace", "old1", "build_go_fetch", delivery.StepRunning, boot.Add(-time.Hour))
+	put("ace", "mjk3l9x0abcdef", "check ask", delivery.StepRunning, boot.Add(-time.Hour))
+	put("ace", "mjk3l9x0abcdee", "check ask", delivery.StepRunning, time.Now().Add(-5*time.Minute))
+	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) != 2 || ended[0] == ended[1] {
+		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)
+	}
+	if s := k.last("ace.mjk3l9x0abcdef"); s.State != delivery.StepFailed || !strings.HasPrefix(s.Reason, "past the step bound: its asker stopped waiting at ") {
+		t.Errorf("the old ask = %+v", s)
+	}
+	for key, state := range map[string]string{"ace.mjk3l9x0abcdee": 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, 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)
+		}
+	}
+	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, "ace", "ace.old2", "stuck for good", nil); err != nil {
+		t.Fatal(err)
+	}
+	if s := k.last("ace.old2"); s.State != delivery.StepFailed || s.Reason != "stuck for good" {
+		t.Errorf("old2 = %+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)
+	}
+}

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) {
+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()}
+	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

go/delivery/stepasks.go

@@ -651,6 +651,11 @@ func ParseAskStepKey(key string) (node, askID string, err error) {
 	return node, askID, nil
 }
 
+// StepBound is how long an asker waits for one ask's step once a build node took it (30 minutes): past it the asker has
+// stopped waiting, so a key still running is nobody's. The build modules end such a key (buildstep.EndStale); the
+// checker's, the builder's and the watchman's own copies of the number are to read this one (issue 1916).
+const StepBound = 30 * time.Minute
+
 // PausesState is the state a build module keeps its nodes' pause keys in. It has no age, unlike StepsState, whose keys
 // are per step and dropped after their age: a pause kept among the steps would end by itself.
 const PausesState = "pauses"

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 {