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 . and its tool is " 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. 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), and 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 || !step.Started.Before(before) || isAsk(k, step) { continue } if err := endRunning(scan, k, step, rev, RestartReason); 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 by a previous process and is ended", k, step.Tool) } ended = append(ended, step.ID) } return ended, nil } // EndOne ends this node's running step id (or the key .) 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. A step not running, or one this module runs now, is refused. " + "id is the step id or the key ..", 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 ."}, "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): } } }