mesh-sdk/go/buildstep/stale.go

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