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

@@ -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()

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 <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) {
+	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 <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, put, node, id, why, live)
+		}}
+}

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

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

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 {