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

@@ -47,6 +47,7 @@ func ScanOf(s stdio.KeptState) Scan {
 // 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")
@@ -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 <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()
@@ -117,17 +120,26 @@ func EndStale(scan Scan, node string, before time.Time, logf func(string, ...any
 			continue
 		}
 		step, err := delivery.DecodeStep(raw)
-		if err != nil || step.State != delivery.StepRunning || !step.Started.Before(before) || isAsk(k, step) {
+		if err != nil || step.State != delivery.StepRunning {
 			continue
 		}
-		if err := endRunning(scan, k, step, rev, RestartReason); err != nil {
+		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 by a previous process and is ended", k, step.Tool)
+			logf("the step %s (%s) was left running and is ended: %s", k, step.Tool, why)
 		}
 		ended = append(ended, step.ID)
 	}
@@ -184,7 +196,8 @@ func EndOne(scan Scan, node, id, why string, live func(id string) bool) (deliver
 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. " +
+			"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>"},
@@ -243,3 +256,17 @@ func EndStaleAtStart(ctx context.Context, scan Scan, node string, before time.Ti
 		}
 	}
 }
+
+// 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

@@ -3,6 +3,7 @@ package buildstep
 import (
 	"encoding/json"
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"strings"
 	"testing"
 	"time"
 
@@ -47,8 +48,8 @@ func scanOf(k *keys, race func(key string)) Scan {
 	}
 }
 
-// 1890: EndStale ends only this node's hand-started steps left running before the process began, never a pulled ask's
+// 1890: EndStale ends only this node's hand-started steps left running before the process began and the asks past the
-// (the stream redelivers it) and never a key written meanwhile; EndOne ends one by hand, never one this process started.
+// 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()
@@ -61,6 +62,7 @@ func TestEndStaleAndEndOne(t *testing.T) {
 	}
 	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))
@@ -74,13 +76,16 @@ func TestEndStaleAndEndOne(t *testing.T) {
 		}
 	})
 	ended, err := EndStale(scan, "ace", boot, nil)
-	if err != nil || len(ended) != 1 || ended[0] != "old1" {
+	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)
 	}
-	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("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)
 		}

go/delivery/stepasks.go

@@ -535,6 +535,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"