mesh/delivery covered after its failure: a later walk that contains it was delivered: plan-1792794944815912745 (novox/mesh-sdk at e1f393bb into main, don…
mesh/check pass 403a5d72ab93 on 1ff6d339780c | unit 4545 run, 2097 kept, 13 skipped | integration 4595 run, 2188 kept, 21 skipped | over bound: unit, …
mesh/delivery-group group fix/1930-digest-shape merged: every member is merged
A check ask's answer kept the digest's packages as a map keyed by Go import path;
the node runtime refuses a state value with a field ending in a credential word,
so the end of any check of mesh-controller (internal/token) was never written and
the ask was acked anyway. The level's digest is now delivery.LevelDigest, packages
a sorted list of {package, digest}, reading the old map form too. finish logs an
end refused every time, writes a small failed end naming the write's error, and
the puller gives the ask back instead of acking it when no end could be written.
147 lines
5.4 KiB
Go
147 lines
5.4 KiB
Go
// Package buildstep is what both 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) and the fetch of one commit over SSH (issue 1474).
|
|
package buildstep
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"strconv"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// Put writes one key of a state: stdio.State(delivery.StepsState).Put's shape (go/state.go).
|
|
type Put func(key string, value any) (uint64, 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
|
|
|
|
// NewID is a step id: the time in milliseconds in base 36 and six random hex digits, lower-case letters and
|
|
// digits only, so it is one key token and differs between two steps of one node even in one millisecond.
|
|
func NewID() string {
|
|
var b [3]byte
|
|
_, _ = rand.Read(b[:])
|
|
return strconv.FormatInt(now().UnixMilli(), 36) + hex.EncodeToString(b[:])
|
|
}
|
|
|
|
// 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, NewID())
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
go end(ctx, put, key, step, run)
|
|
return step.ID, nil
|
|
}
|
|
|
|
// begin writes the step running under id.
|
|
func begin(put Put, node, tool, id string) (delivery.Step, string, error) {
|
|
step := delivery.Step{ID: id, Tool: tool, Node: node, State: delivery.StepRunning, Started: now().UTC()}
|
|
key, err := delivery.StepKey(node, step.ID)
|
|
if err != nil {
|
|
return delivery.Step{}, "", err
|
|
}
|
|
if err := write(put, key, step); err != nil {
|
|
return delivery.Step{}, "", fmt.Errorf("buildstep: 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 delivery.Step, run func(ctx context.Context) (any, error)) {
|
|
result, err := protect(ctx, run)
|
|
_ = finish(put, 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). 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, key string, step delivery.Step, result any, err error, logf func(format string, args ...any)) error {
|
|
step.Ended = now().UTC()
|
|
switch ref := (*Refusal)(nil); {
|
|
case err == nil:
|
|
step.State = delivery.StepPassed
|
|
if result != nil {
|
|
raw, merr := json.Marshal(result)
|
|
if merr != nil {
|
|
step.State, step.Reason = delivery.StepFailed, "the result is not JSON: "+merr.Error()
|
|
} else {
|
|
step.Result = raw
|
|
}
|
|
}
|
|
case errors.As(err, &ref):
|
|
step.State, step.Reason = delivery.StepRefused, err.Error()
|
|
default:
|
|
step.State, step.Reason = delivery.StepFailed, err.Error()
|
|
}
|
|
werr := writeEnd(put, key, step)
|
|
if werr == nil {
|
|
return 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 = delivery.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 fmt.Errorf("buildstep: no end of the step %s could be written: %w", key, ferr)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// writeEnd writes the step's end, trying EndAttempts times.
|
|
func writeEnd(put Put, key string, step delivery.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 delivery.Step) error {
|
|
raw, err := delivery.EncodeStep(step)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = put(key, json.RawMessage(raw))
|
|
return err
|
|
}
|