Merge pull request 'A build step names the process that runs it, so a restarted build module clears what a gone process left running (issue 1890)' (#105) from fix/1890-step-owner into main

go/buildstep/conformance/conformance.go

@@ -4,7 +4,9 @@
 // path, the code that turns a runner's output into the failures of its test result; the kit fails the test when a
 // failure is not named, its cause is missing or wrong, or a passing line stands in its place.
 //
-// A catalogue check (mesh-catalog checks/failurekit) refuses a build module with a test step whose tests do not call Run.
+// A catalogue check (mesh-catalog checks/failurekit) refuses a build module with a test step whose tests do not call Run,
+// and any build module whose tests do not call LeftRunning, the kit's case of steps a previous process left running
+// (leftrunning.go, issue 1890).
 package conformance
 
 import (

go/buildstep/conformance/conformance_test.go

@@ -1,9 +1,11 @@
 package conformance
 
 import (
+	"context"
 	"encoding/json"
 	"strings"
 	"testing"
+	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/buildstep"
 )
@@ -19,6 +21,23 @@ func TestTheFailureContract(t *testing.T) {
 		})
 	})
 
+	// 1890/AC7: the SDK's start sweep keeps the left-running case, and the kit refuses a sweep that ends nothing or ends
+	// everything.
+	t.Run("the left-running case", func(t *testing.T) {
+		LeftRunning(t, func(ctx context.Context, scan buildstep.Scan, node string, before time.Time, logf func(string, ...any)) {
+			buildstep.EndStaleAtStart(ctx, scan, node, before, logf)
+		})
+		if p := JudgeLeftRunning(func(context.Context, buildstep.Scan, string, time.Time, func(string, ...any)) {}); len(p) != 11 {
+			t.Errorf("a sweep that ends nothing: %d problems, want 11 (three steps not ended and unsaid, one owner unnamed, two asks not deleted and unsaid):\n%s", len(p), strings.Join(p, "\n"))
+		}
+		wrongNode := func(ctx context.Context, scan buildstep.Scan, _ string, before time.Time, logf func(string, ...any)) {
+			buildstep.EndStaleAtStart(ctx, scan, "kitother", before, logf)
+		}
+		if p := JudgeLeftRunning(wrongNode); len(p) == 0 {
+			t.Error("a sweep of another node's steps is not refused")
+		}
+	})
+
 	// node's output cut to its first fifteen lines with no cause, whose first line was a passing test of the first file
 	// (mesh-build-node's lines() before the contract), and a Go failure keeping its first line.
 	t.Run("the kit refuses a failure that hides its cause", func(t *testing.T) {

go/buildstep/conformance/leftrunning.go

@@ -0,0 +1,212 @@
+package conformance
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/buildstep"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// **A step left running by a previous process** (novox/hq issue 1890, note 21): when a delivery replaced mesh-build-go
+// on a node while a check ask ran, the new process's start sweep let the ask's key stand as running, and it held one
+// of the node's build slots until a hand end. Every build module runs this case (LeftRunning) through the start sweep
+// its main runs: the hand-started steps a previous process left are ended at once and loud, its asks' keys deleted (the
+// stream redelivers the ask, novox/hq ADR 0525 §6, and its asker keeps waiting for that), what this process runs kept. A
+// catalogue check (mesh-catalog checks/failurekit) refuses a build module whose tests do not call LeftRunning.
+
+// Sweep is a build module's own start sweep, as its main runs it once at its start: it ends node's steps in scan that
+// no process of the module runs any more (buildstep.EndStaleAtStart, here with the runtime already serving), before
+// being the start of the process, saying each end with logf.
+type Sweep func(ctx context.Context, scan buildstep.Scan, node string, before time.Time, logf func(string, ...any))
+
+// LeftNode is the node the case's steps are of.
+const LeftNode = "kitnode"
+
+// leftCase is one step in the steps state before the sweep, and whether the sweep must end it.
+type leftCase struct {
+	id, tool, owner, state string
+	node                   string
+	age                    time.Duration // started this long before the process began (negative: after it)
+	ended                  bool
+	deleted                bool // a gone process's ask: its key is deleted, not ended
+}
+
+// gone is a previous process's owner on the same node.
+var gone = "b00t-of-the-kit.4242.mvp0000"
+
+func leftCases() []leftCase {
+	return []leftCase{
+		{id: "mvpwlkzi7e6ca8", tool: "check ask", owner: gone, age: 5 * time.Minute, deleted: true}, // note 21's ghost
+		{id: "mvpwlkzi7e6ca9", tool: "delivery ask", owner: gone, age: -time.Second, deleted: true},
+		{id: "gonehand1", tool: "build_x_test", owner: gone, age: time.Minute, ended: true},
+		{id: "oldhand1", tool: "build_x_fetch", age: time.Hour, ended: true},                // no owner: an older build's
+		{id: "mvpwlkzi7e6caa", tool: "check ask", age: time.Hour, ended: true},              // no owner, past the bound
+		{id: "mvpwlkzi7e6cab", tool: "check ask", age: 5 * time.Minute},                     // no owner, the stream's still
+		{id: "mvpwlkzi7e6cac", tool: "check ask", owner: buildstep.Owner, age: time.Minute}, // this process's
+		{id: "minehand1", tool: "build_x_test", owner: buildstep.Owner, age: -time.Second},
+		{id: "donehand1", tool: "build_x_test", owner: gone, age: time.Hour, state: delivery.StepPassed},
+		{id: "otherhand1", tool: "build_x_test", owner: gone, age: time.Hour, node: "kitother"},
+	}
+}
+
+// LeftRunning runs the case through sweep and fails the test for each step the sweep ended or kept wrongly, or ended
+// without saying it.
+func LeftRunning(t *testing.T, sweep Sweep) {
+	t.Helper()
+	for _, problem := range JudgeLeftRunning(sweep) {
+		t.Error(problem)
+	}
+}
+
+// JudgeLeftRunning says what is wrong with what sweep did to the case; nothing when it keeps the rule.
+func JudgeLeftRunning(sweep Sweep) []string {
+	st := &steps{vals: map[string][]byte{}, revs: map[string]uint64{}}
+	before := time.Now()
+	for _, c := range leftCases() {
+		node, state := c.node, c.state
+		if node == "" {
+			node = LeftNode
+		}
+		if state == "" {
+			state = delivery.StepRunning
+		}
+		key, _ := delivery.StepKey(node, c.id)
+		raw, err := delivery.EncodeStep(delivery.Step{ID: c.id, Tool: c.tool, Node: node, State: state, Started: before.Add(-c.age).UTC(), Owner: c.owner})
+		if err != nil {
+			return []string{"the kit's step " + c.id + ": " + err.Error()}
+		}
+		st.set(key, raw)
+	}
+	served := make(chan struct{})
+	close(served)
+	was := buildstep.StaleServed
+	buildstep.StaleServed = func() <-chan struct{} { return served }
+	defer func() { buildstep.StaleServed = was }()
+	var mu sync.Mutex
+	var said []string
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	sweep(ctx, st.scan(), LeftNode, before, func(f string, a ...any) {
+		mu.Lock()
+		defer mu.Unlock()
+		said = append(said, fmt.Sprintf(f, a...))
+	})
+	mu.Lock()
+	log := strings.Join(said, "\n")
+	mu.Unlock()
+	var problems []string
+	for _, c := range leftCases() {
+		node := c.node
+		if node == "" {
+			node = LeftNode
+		}
+		key, _ := delivery.StepKey(node, c.id)
+		raw := st.get(key)
+		if c.deleted {
+			if raw != nil {
+				problems = append(problems, fmt.Sprintf("left running: the ask %s of the gone owner %q is still held after the sweep (%s); "+
+					"its key is deleted, never ended: the stream redelivers it (ADR 0525 §6)", key, c.owner, raw))
+			}
+			if !strings.Contains(log, key) || !strings.Contains(log, c.owner) {
+				problems = append(problems, "left running: the delete of "+key+" is not said with its owner "+c.owner)
+			}
+			continue
+		}
+		if raw == nil {
+			problems = append(problems, "left running: the step "+key+" is gone after the sweep, want it kept or ended")
+			continue
+		}
+		s, err := delivery.DecodeStep(raw)
+		if err != nil {
+			problems = append(problems, key+": "+err.Error())
+			continue
+		}
+		want := c.state
+		if want == "" {
+			want = delivery.StepRunning
+		}
+		if c.ended {
+			want = delivery.StepFailed
+		}
+		if s.State != want {
+			problems = append(problems, fmt.Sprintf("left running: the step %s (%s, owner %q, %s before the start) is %s after the sweep, want %s", key, c.tool, c.owner, c.age, s.State, want))
+		}
+		if c.ended && !strings.Contains(log, key) {
+			problems = append(problems, "left running: the end of "+key+" is not said")
+		}
+		if c.ended && c.owner != "" && !strings.Contains(log, c.owner) {
+			problems = append(problems, "left running: the end of "+key+" does not name its owner "+c.owner)
+		}
+	}
+	sort.Strings(problems)
+	return problems
+}
+
+// steps is a steps state with revisions, whose Update lands only at the revision read (stdio.ErrChanged otherwise).
+type steps struct {
+	mu   sync.Mutex
+	vals map[string][]byte
+	revs map[string]uint64
+}
+
+func (s *steps) set(key string, raw []byte) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.vals[key], s.revs[key] = raw, s.revs[key]+1
+}
+
+func (s *steps) get(key string) []byte {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.vals[key]
+}
+
+func (s *steps) scan() buildstep.Scan {
+	return buildstep.Scan{
+		Keys: func() ([]string, error) {
+			s.mu.Lock()
+			defer s.mu.Unlock()
+			var out []string
+			for k := range s.vals {
+				out = append(out, k)
+			}
+			return out, nil
+		},
+		Get: func(key string) (json.RawMessage, uint64, error) {
+			s.mu.Lock()
+			defer s.mu.Unlock()
+			return s.vals[key], s.revs[key], nil
+		},
+		Delete: func(key string, revision uint64) error {
+			s.mu.Lock()
+			defer s.mu.Unlock()
+			if s.revs[key] != revision {
+				return stdio.ErrChanged
+			}
+			delete(s.vals, key)
+			s.revs[key]++
+			return nil
+		},
+		Update: func(key string, value any, revision uint64) (uint64, error) {
+			raw, err := json.Marshal(value)
+			if err != nil {
+				return 0, err
+			}
+			s.mu.Lock()
+			defer s.mu.Unlock()
+			if s.revs[key] != revision {
+				return 0, stdio.ErrChanged
+			}
+			s.vals[key], s.revs[key] = raw, revision+1
+			return revision + 1, nil
+		},
+	}
+}

go/buildstep/pull_test.go

@@ -166,7 +166,7 @@ func TestPullHoldsAtMostItsCapacity(t *testing.T) {
 	waitFor(t, "a phase on the step key", func() bool {
 		key, _ := delivery.AskStepKey("shanks", p.Running()[0].ID)
 		s, _ := st.get(key)
-		return s.State == delivery.StepRunning && s.Phase == "unit: tests of 3 packages" && !s.PhaseAt.IsZero()
+		return s.State == delivery.StepRunning && s.Phase == "unit: tests of 3 packages" && !s.PhaseAt.IsZero() && s.Owner == Owner // 1890/AC5
 	})
 	release <- struct{}{}
 	waitFor(t, "the third taken", func() bool { return src.Waiting(delivery.AskDelivery) == 0 })
@@ -176,7 +176,7 @@ func TestPullHoldsAtMostItsCapacity(t *testing.T) {
 	waitFor(t, "all acked", func() bool { a, _, _ := src.Fates(); return len(a) == 3 })
 	for _, a := range asks {
 		key, _ := delivery.AskStepKey("shanks", a.ID)
-		if s, _ := st.get(key); s.State != delivery.StepPassed || s.Phase != "unit: tests of 3 packages" {
+		if s, _ := st.get(key); s.State != delivery.StepPassed || s.Phase != "unit: tests of 3 packages" || s.Owner != Owner {
 			t.Errorf("the end of %s = %+v, want passed in its last phase", a.ID, s)
 		}
 	}

go/buildstep/stale.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -14,40 +16,82 @@ import (
 
 // **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
+// ask comes back. At its start a module ends the steps its node left running by another process (EndStale: every step
-// its step-end tool ends one by hand (EndOne).
+// names its owner, delivery.Step.Owner; note 21), and its step-end tool ends one by hand (EndOne). The conformance kit's
+// left-running case holds every build module to the start sweep (conformance.LeftRunning).
 
 // 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
+// Boot is when this process began. A step with no owner (written by an older build) started at or after it is taken
-// EndOne refuses it.
+// for this process's, which may still run it, so EndOne refuses it.
 var Boot = time.Now()
 
+// Owner is this process as a step's owner (delivery.Step.Owner): `<boot id>.<pid>.<start>`, the node's boot id
+// (/proc/sys/kernel/random/boot_id), the pid and Boot in milliseconds, base 36. The start keeps two processes apart
+// that a reboot or a pid namespace gave the same pid. A running step whose owner is another is a previous process's.
+var Owner = ownerOf(bootID, os.Getpid(), Boot)
+
+// bootID is the node's boot id as Owner holds it; noBootID when it could not be read, which EndStaleAtStart says once.
+var bootID = readBootID()
+
+const noBootID = "no-boot-id"
+
+func ownerOf(bootID string, pid int, start time.Time) string {
+	return bootID + "." + strconv.Itoa(pid) + "." + strconv.FormatInt(start.UnixMilli(), 36)
+}
+
+// readBootID is the node's boot id, or noBootID where it cannot be read (the pid and the start then tell the processes
+// apart).
+func readBootID() string {
+	b, err := os.ReadFile("/proc/sys/kernel/random/boot_id")
+	if id := strings.TrimSpace(string(b)); err == nil && id != "" {
+		return id
+	}
+	return noBootID
+}
+
 // 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
+// none), a write that lands only while the key is still at the revision read (KeptState.Update), and a delete of a key
-// stdio.ErrChanged when another writer was first.
+// still at the revision read; both fail 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)
+	Delete func(key string, revision uint64) error
 }
 
 // ScanOf is the Scan of a state the process keeps.
+//
+// Its Delete reads the key again and deletes it only while it is still at the revision read: the runtime's
+// mesh/state.delete takes no revision yet, so a write landing between that read and the delete (a few milliseconds) is
+// deleted with it. For a step key that is a redelivered ask begun on this node in that moment: its running key leaves
+// the steps state until its run writes its end, which its asker still reads (a watch takes puts only).
 func ScanOf(s stdio.KeptState) Scan {
-	return Scan{Keys: s.Keys, Update: s.Update, Get: func(key string) (json.RawMessage, uint64, error) {
+	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
+	}
+	return Scan{Keys: s.Keys, Update: s.Update, Get: get, Delete: func(key string, revision uint64) error {
+		_, rev, err := get(key)
+		if err != nil {
+			return err
+		}
+		if rev != revision {
+			return stdio.ErrChanged
+		}
+		return s.Delete(key)
 	}}
 }
 
 // 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.
+// A key a gone process owns is deleted, never ended (EndStale); one with no owner is ended only once the asker's bound
+// has passed (delivery.StepBound), when it is nobody's.
 func isAsk(key string, s delivery.Step) bool {
 	_, _, err := delivery.ParseAskStepKey(key)
 	return err == nil && strings.HasSuffix(s.Tool, " ask")
@@ -95,11 +139,23 @@ func endRunning(scan Scan, key string, step delivery.Step, revision uint64, why
 	return err
 }
 
-// EndStale ends this node's steps that are running and started before `before` (the start of this process) as failed
+// EndStale ends this node's running steps that no process of this module runs any more, and answers their ids. Only keys
-// with RestartReason, and answers their ids. Only keys of node are read; a value that is no step is left alone. A step
+// of node are read; a value that is no step is left alone.
-// of a pulled ask is left to the stream's redelivery (isAsk) until its start is older than delivery.StepBound: its asker
+//   - A hand-started step with an owner (delivery.Step.Owner) that is not this process (Owner) is ended failed with
-// stopped waiting then, whether the stream gave it to another node or gave it up, and it is ended as failed, "past the
+//     RestartReason at once, whatever its age: its process is gone and nothing redelivers it, so nothing else would
-// step bound: its asker stopped waiting at <time>". A key written by another writer since it was read
+//     write its end (issue 1890, note 21).
+//   - A pulled ask's step with an owner that is not this process is deleted, not ended (novox/hq ADR 0525 §6): the
+//     stream redelivers an ask whose node stopped (the puller neither ends nor acknowledges one at its stop), and its
+//     asker takes the first end it reads, so an end written here would fail a check the redelivery would still run. Its
+//     asker's watch takes puts only and keeps waiting for the redelivery's key, its bound the backstop; the stale running
+//     key leaves the steps state, and with it the delivery view's and the checker's count of what runs (note 21: it
+//     held one of shanks' two slots).
+//   - A step owned by this process is left: it may run now. A redelivered ask begun on this node is owned by it.
+//   - A step with no owner (written by an older build) keeps the earlier rule: a hand-started step started before
+//     `before` (the start of this process) is ended; a pulled ask is left to the stream's redelivery (isAsk) until its
+//     start is older than delivery.StepBound, and then ended "past the step bound: its asker stopped waiting at <time>".
+//
+// Each end and delete is logged, naming the step, its owner and why. 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()
@@ -123,32 +179,59 @@ func EndStale(scan Scan, node string, before time.Time, logf func(string, ...any
 		if err != nil || step.State != delivery.StepRunning {
 			continue
 		}
-		why := RestartReason
+		why, said := staleWhy(k, step, before)
-		if isAsk(k, step) {
+		if why == "" {
-			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")
+		done := "ended"
-		} else if !step.Started.Before(before) {
+		if step.Owner != "" && isAsk(k, step) {
-			continue
+			done = "deleted, left to the stream's redelivery"
+			if scan.Delete == nil {
+				err = fmt.Errorf("the steps state takes no delete")
+			} else {
+				err = scan.Delete(k, rev)
 			}
-		if err := endRunning(scan, k, step, rev, why); err != nil {
+		} else {
+			err = endRunning(scan, k, step, rev, why)
+		}
+		if 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)
+			logf("the step %s (%s, started %s) was left running and is %s: %s; %s", k, step.Tool,
+				step.Started.UTC().Format(time.RFC3339), done, why, said)
 		}
 		ended = append(ended, step.ID)
 	}
 	return ended, nil
 }
 
+// staleWhy is why a running step is ended (or, for an ask with an owner, deleted) by EndStale and what the log says of
+// its owner, or "" to leave it.
+func staleWhy(key string, step delivery.Step, before time.Time) (why, said string) {
+	switch {
+	case step.Owner == Owner:
+		return "", ""
+	case step.Owner != "":
+		return RestartReason, "its owner " + step.Owner + " is not this process " + Owner
+	case isAsk(key, step):
+		if !step.Started.Add(delivery.StepBound).Before(now()) {
+			return "", ""
+		}
+		return "past the step bound: its asker stopped waiting at " + step.Started.Add(delivery.StepBound).UTC().Format("2006-01-02T15:04:05Z"),
+			"it names no owner (an older build wrote it)"
+	case step.Started.Before(before):
+		return RestartReason, "it names no owner (an older build wrote it) and started before this process"
+	}
+	return "", ""
+}
+
 // 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
+// refused, and so is one this process owns (Owner; for a step with no owner, one started since Boot: it may still run,
-// tool started), one live reports as held, and one whose key changed since it was read.
+// 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 {
@@ -178,7 +261,7 @@ func EndOne(scan Scan, node, id, why string, live func(id string) bool) (deliver
 	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)) {
+	if step.Owner == Owner || (step.Owner == "" && !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 {
@@ -226,14 +309,24 @@ var (
 	StaleRetry    = 5 * time.Second
 )
 
+// StaleServed is what EndStaleAtStart waits for before it reads the steps state: the runtime serving (stdio.Served).
+// The conformance kit's left-running case (conformance.LeftRunning) hands it a closed channel, to run a module's own
+// start sweep in its test.
+var StaleServed = stdio.Served
+
 // 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.
+// then ends this node's steps a previous process left running (EndStale), trying again when the state cannot be read
-// A step this process starts is never one of them: it starts after `before`.
+// yet. A step names its owner, so `before` (the start of this process) matters only for a step with no owner, written
+// by an older build: such a step started before it is a previous process's. When the node's boot id cannot be read it
+// says so once: Owner then tells processes apart by their pid and start only.
 func EndStaleAtStart(ctx context.Context, scan Scan, node string, before time.Time, logf func(string, ...any)) {
 	select {
 	case <-ctx.Done():
 		return
-	case <-stdio.Served():
+	case <-StaleServed():
+	}
+	if bootID == noBootID && logf != nil {
+		logf("the boot id of %s cannot be read (/proc/sys/kernel/random/boot_id): its steps' owner is %s, told apart by pid and start only", node, Owner)
 	}
 	for attempt := 1; ; attempt++ {
 		ended, err := EndStale(scan, node, before, logf)

go/buildstep/stale_test.go

@@ -2,6 +2,7 @@ package buildstep
 
 import (
 	"encoding/json"
+	"fmt"
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"strings"
 	"testing"
@@ -33,6 +34,15 @@ func scanOf(k *keys, race func(key string)) Scan {
 			raw, err := delivery.EncodeStep(s[len(s)-1])
 			return raw, uint64(len(s)), err
 		},
+		Delete: func(key string, revision uint64) error {
+			k.mu.Lock()
+			defer k.mu.Unlock()
+			if uint64(len(k.puts[key])) != revision {
+				return stdio.ErrChanged
+			}
+			delete(k.puts, key)
+			return nil
+		},
 		Update: func(key string, value any, revision uint64) (uint64, error) {
 			if race != nil {
 				race(key)
@@ -48,65 +58,85 @@ 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 and the asks past the
+// 1890 and 1890/AC6: EndStale ends this node's hand-started running steps of another owner at once, whatever their age,
-// 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.
+// and deletes its asks of another owner (the stream redelivers them, ADR 0525 §6); of
+// the steps with no owner (an older build's), the hand-started ones left running before the process began and the asks
+// past the step bound (a younger ask is the stream's to redeliver); never a step this process owns or a key written
+// meanwhile. EndOne ends one by hand, never one this process owns or 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) {
+	gone := ownerOf("b00t", 4242, boot.Add(-time.Hour))
+	put := func(node, id, tool, state string, started time.Time, owner string) {
 		key, _ := delivery.StepKey(node, id)
-		_, _ = k.put(key, json.RawMessage(mustStep(t, delivery.Step{ID: id, Tool: tool, 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, Owner: owner})))
 	}
-	put("ace", "old1", "build_go_fetch", delivery.StepRunning, boot.Add(-time.Hour))
+	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", "mjk3l9x0abcdef", "check ask", delivery.StepRunning, boot.Add(-time.Hour), "")
-	put("ace", "mjk3l9x0abcdee", "check ask", delivery.StepRunning, time.Now().Add(-5*time.Minute))
+	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", "mjk3l9x0abcded", "check ask", delivery.StepRunning, time.Now().Add(-5*time.Minute), gone) // note 21's ghost
-	put("ace", "new1", "build_go_fetch", delivery.StepRunning, boot.Add(time.Second))
+	put("ace", "gone2", "build_go_test", delivery.StepRunning, boot.Add(time.Second), gone)
-	put("ace", "done1", "build_go_fetch", delivery.StepPassed, boot.Add(-time.Hour))
+	put("ace", "mine1", "check ask", delivery.StepRunning, boot.Add(-time.Hour), Owner)
-	put("g14", "other1", "build_go_fetch", 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), gone)
+	put("g14", "other1", "build_go_fetch", delivery.StepRunning, boot.Add(-time.Hour), gone)
 	// 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))
+			put("ace", "raced1", "build_go_test", delivery.StepPassed, boot.Add(-time.Hour), "")
 		}
 	})
-	ended, err := EndStale(scan, "ace", boot, nil)
+	var said []string
-	if err != nil || len(ended) != 2 || ended[0] == ended[1] {
+	ended, err := EndStale(scan, "ace", boot, func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) })
+	if err != nil || len(ended) != 4 {
 		t.Fatalf("ended %v, %v", ended, err)
 	}
-	if s := k.last("ace.old1"); s.State != delivery.StepFailed || s.Reason != RestartReason || s.Ended.IsZero() {
+	if _, held := k.puts["ace.mjk3l9x0abcded"]; held {
-		t.Errorf("old1 = %+v", s)
+		t.Errorf("the ghost ask's key is still held: %+v", k.last("ace.mjk3l9x0abcded"))
+	}
+	for _, key := range []string{"ace.old1", "ace.gone2"} {
+		if s := k.last(key); s.State != delivery.StepFailed || s.Reason != RestartReason || s.Ended.IsZero() {
+			t.Errorf("%s = %+v", key, 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 log := strings.Join(said, "\n"); !strings.Contains(log, "ace.mjk3l9x0abcded") || !strings.Contains(log, "deleted, left to the stream's redelivery") || !strings.Contains(log, "its owner "+gone+" is not this process "+Owner) {
+		t.Errorf("the log does not name the ghost and its owner:\n%s", log)
+	}
+	for key, state := range map[string]string{"ace.mjk3l9x0abcdee": delivery.StepRunning, "ace.mine1": 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
+	// EndOne: refuses without why, a held step, one this process owns or (with no owner) started, a step not running and
-	// running and another node's; ends a step a previous process left.
+	// 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"}} {
+	for _, c := range []struct{ id, why string }{{"old1", ""}, {"new1", "stuck"}, {"mine1", "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))
+	put("ace", "old2", "build_go_fetch", delivery.StepRunning, boot.Add(time.Second), gone)
 	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" {
+	if s := k.last("ace.old2"); s.State != delivery.StepFailed || s.Reason != "stuck for good" || s.Owner != gone {
 		t.Errorf("old2 = %+v", s)
 	}
+	if a, b := ownerOf("b", 1, boot), ownerOf("b", 1, boot.Add(time.Second)); a == b {
+		t.Errorf("two starts of one pid share the owner %q", a)
+	}
 }
 
 func mustStep(t *testing.T, s delivery.Step) []byte {

go/buildstep/step.go

@@ -52,9 +52,9 @@ func Start(ctx context.Context, put Put, node, tool string, run func(ctx context
 	return step.ID, nil
 }
 
-// begin writes the step running under id.
+// begin writes the step running under id, owned by this process (Owner).
 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}
+	step := delivery.Step{ID: id, Tool: tool, Node: node, State: delivery.StepRunning, Started: now().UTC(), What: what, Owner: Owner}
 	key, err := delivery.StepKey(node, step.ID)
 	if err != nil {
 		return delivery.Step{}, "", err

go/buildstep/step_test.go

@@ -70,8 +70,9 @@ func TestStartAnswersAtOnceWithTheStepRunning(t *testing.T) {
 		t.Errorf("Start took %v", d)
 	}
 	s := k.last("ace." + id)
-	if s.State != delivery.StepRunning || s.Tool != "fetch" || s.Node != "ace" || s.ID != id || s.Started.IsZero() {
+	// 1890/AC5: the running key names this process as its owner.
-		t.Errorf("step = %+v", s)
+	if s.State != delivery.StepRunning || s.Tool != "fetch" || s.Node != "ace" || s.ID != id || s.Started.IsZero() || s.Owner != Owner || Owner == "" {
+		t.Errorf("step = %+v, want owner %q", s, Owner)
 	}
 }
 
@@ -100,7 +101,7 @@ func TestStepEndsPassedFailedOrRefusedOnItsKey(t *testing.T) {
 				t.Fatalf("ended key %q", got)
 			}
 			s := k.last("ace." + id)
-			if s.State != c.state || s.Reason != c.reason || string(s.Result) != c.result || s.Ended.IsZero() {
+			if s.State != c.state || s.Reason != c.reason || string(s.Result) != c.result || s.Ended.IsZero() || s.Owner != Owner {
 				t.Errorf("step = %+v", s)
 			}
 		})

go/delivery/steps.go

@@ -78,6 +78,10 @@ type Step struct {
 	// A reader waiting for the end treats such a write as any running one.
 	Phase   string    `json:"phase,omitempty"`
 	PhaseAt time.Time `json:"phase-at,omitzero"`
+	// Owner is the process that runs the step: its node's boot id, its pid and its start (buildstep.Owner), written when
+	// the step begins and kept by every later write of it. A build module's start sweep ends at once a running step whose
+	// owner is not itself: that process is gone (issue 1890, note 21). Empty in a key an older build wrote.
+	Owner string `json:"owner,omitempty"`
 }
 
 func (s Step) check() error {