mesh-delivery/modules/mesh-delivery-sdk/steprun/step.go

// Package steprun is what the build modules share: the runner of a long step that answers a tool call at once
// and ends on its key in the module's `steps` state (novox/hq ADR 0469 §2; issue 1471), the toolchain guard
// (issue 1473), the fetch of one commit over SSH (issue 1474), and the failure contract of a test level with the parsers
// of go test -json and node --test's TAP that fill it (failure.go, issue 1901; its conformance kit is package conformance).
package steprun

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"time"

	buildrun "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildrun"
	sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
	"git.novox.be/novox/mesh-sdk/go/kvlimit"
)

// Put writes one key of a state: the steps state.Put's shape (go/state.go).
type Put func(key string, value any) (uint64, error)

// Keep keeps a step's answer that is over what the steps state holds per value as the object named object in the build
// module's outputs store (design 47 G A-290), the object the seat's verb output answers: Puller.KeepOutput, KeepOutputIn.
type Keep func(object string, body []byte) error

// EndAttempts and EndRetry bound how a step's end is written: a write that fails is tried again, since the end
// is the only way the step's reader learns it is over.
var (
	EndAttempts = 5
	EndRetry    = 2 * time.Second
)

var now = time.Now

// Start begins a step of tool on node and answers its id at once: the key <node>.<id> holds state running when
// Start returns. run is then called in a goroutine of its own, under ctx (the process's, not the tool call's: the
// call is cut at 30 s, the step is not). When it returns, the key is written passed with run's result, failed with
// 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, sdkstep.NewID(), "")
	if err != nil {
		return "", err
	}
	go end(ctx, put, key, step, run)
	return step.ID, nil
}

// begin writes the step running under id, owned by this process (Owner).
func begin(put Put, node, tool, id, what string) (sdkstep.Step, string, error) {
	return beginRun(put, "", node, tool, id, what, 0)
}

// beginRun is begin for the run-th run of an ask (the bus's delivery count; 0 for a step that is no ask), keyed under the
// asker its subject names, or main's form for none (design 47 G §481, §483).
func beginRun(put Put, asker, node, tool, id, what string, run int) (sdkstep.Step, string, error) {
	step := sdkstep.Step{ID: id, Tool: tool, Node: node, State: sdkstep.StepRunning, Started: now().UTC(), What: what, Owner: buildrun.Owner, Run: run}
	key, err := sdkstep.StepKeyOf(asker, node, step.ID)
	if err != nil {
		return sdkstep.Step{}, "", err
	}
	if err := write(put, key, step); err != nil {
		return sdkstep.Step{}, "", fmt.Errorf("steprun: writing the running step %s: %w", key, err)
	}
	return step, key, nil
}

// end runs run and writes the step's end.
func end(ctx context.Context, put Put, key string, step sdkstep.Step, run func(ctx context.Context) (any, error)) {
	result, err := protect(ctx, run)
	_, _ = finish(put, nil, key, step, result, err, log.Printf) // finish says a refused end itself
}

// finish writes the step's end from run's result or error: passed, failed or refused (a *Refusal), and answers the end it
// wrote. An end whose value would be over what the steps state holds per value keeps its answer as an object with keep
// (keepOver, design 47 G A-290). A write that fails is tried EndAttempts times. When the whole end is refused every time (a result the state will not take: issue 1930), the
// refusal is logged and a small failed end without the result is written instead, naming the write's error, so the
// step's reader learns it is over and why. finish answers an error only when no end could be written at all, and then
// the step's ask is not acknowledged as done.
func finish(put Put, keep Keep, key string, step sdkstep.Step, result any, err error, logf func(format string, args ...any)) (sdkstep.Step, error) {
	step.Ended = now().UTC()
	switch ref, drop := (*Refusal)(nil), (*Dropped)(nil); {
	case err == nil:
		step.State = sdkstep.StepPassed
		if result != nil {
			raw, merr := json.Marshal(result)
			if merr != nil {
				step.State, step.Reason = sdkstep.StepFailed, "the result is not JSON: "+merr.Error()
			} else {
				step.Result = raw
			}
		}
		if ans, ok := runAnswerIn(result); ok && ans.Exit != 0 && step.State == sdkstep.StepPassed {
			// A run whose program exited other than 0 ends failed, its answer kept, as its step-ended says (design 47 G A-290).
			step.State, step.Reason = sdkstep.StepFailed, exitedWhy(ans)
		}
	case errors.As(err, &drop):
		step.State, step.Reason = sdkstep.StepDropped, err.Error()
	case errors.As(err, &ref):
		step.State, step.Reason = sdkstep.StepRefused, err.Error()
	default:
		step.State, step.Reason = sdkstep.StepFailed, err.Error()
		step.Passing = sdkstep.IsPassing(err) // it ended on a transport fault: its asker may ask it again (ADR 0664 §3)
		// A failed step may answer what it found on the way (a build's out-of-step files): kept on its end.
		if result != nil {
			if raw, merr := json.Marshal(result); merr == nil {
				step.Result = raw
			}
		}
	}
	step = keepOver(keep, step, result)
	werr := writeEnd(put, key, step)
	if werr == nil {
		return step, nil
	}
	if logf != nil {
		logf("the end of the step %s (%s) could not be written in %d tries: %v", key, step.State, EndAttempts, werr)
	}
	ended := step.State
	why := fmt.Sprintf("the step ended %s, and its end could not be written: %v", ended, werr)
	if step.Reason != "" {
		why = fmt.Sprintf("the step ended %s (%s), and its end could not be written: %v", ended, step.Reason, werr)
	}
	step.State, step.Reason, step.Result, step.OutputObject = sdkstep.StepFailed, why, nil, ""
	if ferr := writeEnd(put, key, step); ferr != nil {
		if logf != nil {
			logf("the failed end of the step %s could not be written either: %v", key, ferr)
		}
		return step, fmt.Errorf("steprun: no end of the step %s could be written: %w", key, ferr)
	}
	return step, nil
}

// keepOver is step as its key keeps it (design 47 G A-290): a value within what the steps state holds per value
// (kvlimit.ModuleStateValueBytes less the headers' room) keeps its answer itself; over it, the answer is kept with keep as the
// object sdkstep.OutputObjectName(step.ID), which the key names as its output-object: a run's output, the key keeping the
// run answer without it, else the whole answer, the key's result then empty. A step whose answer cannot be kept ends failed,
// saying why, with no answer, so it never reads passed where its answer is lost.
func keepOver(keep Keep, step sdkstep.Step, result any) sdkstep.Step {
	raw, err := sdkstep.EncodeStep(step)
	if err != nil || len(step.Result) == 0 || kvlimit.Check("steps", "", len(raw), kvlimit.ModuleStateValueBytes) == nil {
		return step
	}
	object, body, rest := sdkstep.OutputObjectName(step.ID), []byte(step.Result), json.RawMessage(nil)
	if ans, ok := runAnswerIn(result); ok {
		body, ans.Output = ans.Output, nil
		if rest, err = json.Marshal(ans); err != nil {
			rest = nil
		}
	}
	lost := func(why string) sdkstep.Step {
		if step.Reason != "" {
			why = step.Reason + "; " + why
		}
		step.State, step.Reason, step.Result, step.Passing = sdkstep.StepFailed, why, nil, false
		return step
	}
	if keep == nil {
		return lost(fmt.Sprintf("the step's answer makes its value %d bytes, over what the steps state holds, and this holder keeps no object for it", len(raw)))
	}
	if err := keep(object, body); err != nil {
		return lost(fmt.Sprintf("the step's answer makes its value %d bytes, over what the steps state holds, and could not be kept as the object %s: %v", len(raw), object, err))
	}
	step.Result, step.OutputObject = rest, object
	return step
}

// exitedWhy is why a run whose program exited other than 0 failed: its exit status and the first line of its standard error.
func exitedWhy(ans sdkstep.RunAnswer) string {
	return fmt.Sprintf("the program exited %d: %s", ans.Exit, ans.Said)
}

// runAnswerIn is the run answer a step handed back, whatever its form (the value or a pointer).
func runAnswerIn(result any) (sdkstep.RunAnswer, bool) {
	switch a := result.(type) {
	case sdkstep.RunAnswer:
		return a, true
	case *sdkstep.RunAnswer:
		if a != nil {
			return *a, true
		}
	}
	return sdkstep.RunAnswer{}, false
}

// again writes the step as given again after a transport fault: the state again with the run's cause, and when the bus gives
// the ask again (ADR 0664 §1). The key is not ended: its asker keeps waiting on it.
func again(put Put, key string, step sdkstep.Step, err error, delay time.Duration) error {
	step.State, step.Reason, step.Passing, step.Result, step.OutputObject = sdkstep.StepAgain, err.Error(), true, nil, ""
	step.Ended, step.AgainAt = time.Time{}, now().UTC().Add(delay)
	return writeEnd(put, key, step)
}

// writeEnd writes the step's end, trying EndAttempts times.
func writeEnd(put Put, key string, step sdkstep.Step) error {
	for attempt := 1; ; attempt++ {
		werr := write(put, key, step)
		if werr == nil || attempt >= EndAttempts {
			return werr
		}
		time.Sleep(EndRetry)
	}
}

// protect is run with a panic turned into the error that ends the step.
func protect(ctx context.Context, run func(ctx context.Context) (any, error)) (result any, err error) {
	defer func() {
		if p := recover(); p != nil {
			result, err = nil, fmt.Errorf("the step panicked: %v", p)
		}
	}()
	return run(ctx)
}

func write(put Put, key string, step sdkstep.Step) error {
	raw, err := sdkstep.EncodeStep(step)
	if err != nil {
		return err
	}
	_, err = put(key, json.RawMessage(raw))
	return err
}