diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index f2e0fa4..960d02e 100644 --- a/go/buildstep/pull.go +++ b/go/buildstep/pull.go @@ -319,7 +319,7 @@ func (p *Puller) next(ctx context.Context, wait time.Duration) (*Pulled, error) // message: the asker puts the ask again under a new id (issue 890). A message whose decline cannot be written is handed // back, as one whose start cannot be. func (p *Puller) decline(msg *Pulled, ask delivery.StepAsk, lacks []string) { - step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID) + step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID, AskWhat(ask)) if err == nil { why := Refuse("the ask %s needs %s, and %s lacks %s: it is left to a node that has it", ask.ID, strings.Join(ask.Needs, ", "), p.Node, strings.Join(lacks, ", ")) @@ -369,7 +369,7 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) { p.decline(msg, ask, lacks) return } - step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID) + step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID, AskWhat(ask)) if err != nil { p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err) _ = msg.Nak() diff --git a/go/buildstep/stale.go b/go/buildstep/stale.go new file mode 100644 index 0000000..4147934 --- /dev/null +++ b/go/buildstep/stale.go @@ -0,0 +1,272 @@ +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. +// 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