mesh-sdk/go/buildstep/stale.go

package buildstep

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strconv"
	"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 by another process (EndStale: every step
// 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 with no owner (written by an older build) started at or after it is taken
// 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), a write that lands only while the key is still at the revision read (KeptState.Update), and a delete of a key
// 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 {
	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.
// 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")
}

// 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 running steps that no process of this module runs any more, and answers their ids. Only keys
// of node are read; a value that is no step is left alone.
//   - A hand-started step with an owner (delivery.Step.Owner) that is not this process (Owner) is ended failed with
//     RestartReason at once, whatever its age: its process is gone and nothing redelivers it, so nothing else would
//     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()
	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, said := staleWhy(k, step, before)
		if why == "" {
			continue
		}
		done := "ended"
		if step.Owner != "" && isAsk(k, step) {
			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)
			}
		} 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, 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 this process owns (Owner; for a step with no owner, one started since 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.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 {
		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
)

// 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 a previous process left running (EndStale), trying again when the state cannot be read
// 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 <-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)
		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)
	}
}