package buildstep import ( "context" "encoding/json" "errors" "fmt" "strings" "time" nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" buildrun "git.novox.be/novox/mesh-delivery/sdk/buildrun" sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep" stdio "git.novox.be/novox/mesh-sdk/go" ) // **The seat's verb drop ends an ask its caller no longer waits for** (novox/hq ADR 0664 ยง5, issue 1748; research 086 // choice 28's cancel). It writes the ask in this node's own pauses state under dropped. (the part build-step), so a // holder that restarts still refuses the ask when the bus gives it again, and ends the step if it runs here: the step's // context is cancelled, its process ended, and its key is written dropped naming who dropped it and why. The state is the // module's, shared by every node: any holder answers the verb, every node refuses the ask when it pulls it (an ask that // waits leaves the stream at its next pull, with no work done), and the node that runs it hears the key by its watch // (watchDrops) and ends the step. // DroppedKeep is how long a dropped ask's key stays: the start deletes older ones. An ask the seat's stream still holds is // younger than its stream's maximum age, which no contract this module reads states, so the bound is a generous week. const DroppedKeep = 7 * 24 * time.Hour // droppedError is why a step's context was cancelled: the seat's verb drop. type droppedError struct{ by, why string } func (d *droppedError) Error() string { by := d.by if by == "" { by = "its asker" } return fmt.Sprintf("the ask was dropped by %s: %s", by, d.why) } // droppedCause is the drop that cancelled ctx, nil when it was none. func droppedCause(ctx context.Context) *droppedError { var d *droppedError if errors.As(context.Cause(ctx), &d) { return d } return nil } // cancellable is ctx with a cancel the verb drop reaches by the ask's id, until the step ends with ctx. func (p *Puller) cancellable(ctx context.Context, id string) context.Context { c, cancel := context.WithCancelCause(ctx) p.mu.Lock() if p.cancels == nil { p.cancels = map[string]context.CancelCauseFunc{} } p.cancels[id] = cancel p.mu.Unlock() return c } // Drop is the seat's verb drop: it keeps the ask's drop in the pauses state and ends the step if it runs here. func (p *Puller) Drop(id, by, why string) error { if strings.TrimSpace(id) == "" || strings.TrimSpace(why) == "" { return errors.New("buildstep: give the ask's id and why it is dropped") } at := now().UTC() raw, err := json.Marshal(nodebuildgo.Drop{V: int64(sdkstep.Version), By: by, Why: why, At: &at}) if err != nil { return err } put := p.PausePut if put == nil { put = p.Put } if _, err := put(sdkstep.DropKey(id), json.RawMessage(raw)); err != nil { return fmt.Errorf("buildstep: writing the drop of %s: %w", id, err) } p.mu.Lock() cancel := p.cancels[id] p.mu.Unlock() if cancel != nil { cancel(&droppedError{by: by, why: why}) } p.Wake() return nil } // watchDrops follows the drops written in the pauses state, by any holder of the seat: one for an ask that runs here ends its // step. An ask that waits is refused when it is pulled (take), on whichever node pulls it. A watch that cannot start is said; // the pull then refuses a dropped ask but cannot end a running one. func (p *Puller) watchDrops() { if p.WatchPauses == nil { p.logf("no watch of the drops in the pauses of %s is set: a dropped ask that runs here is not ended, only refused when it is pulled", p.Node) return } err := p.WatchPauses(sdkstep.DroppedPrefix+"**", func(ch stdio.StateChange) error { id, ok := strings.CutPrefix(ch.Key, sdkstep.DroppedPrefix) if !ok || ch.Op != "put" { return nil } var d nodebuildgo.Drop if json.Unmarshal(ch.Value, &d) != nil { return nil } p.mu.Lock() cancel := p.cancels[id] p.mu.Unlock() if cancel != nil { cancel(&droppedError{by: d.By, why: d.Why}) } return nil }) if err != nil { p.logf("the drops in the pauses of %s cannot be followed: %v; a dropped ask that runs here is not ended", p.Node, err) } } // droppedAsk is the drop kept for the ask, nil for none (or no state to read it from). func (p *Puller) droppedAsk(id string) *nodebuildgo.Drop { if p.Load == nil { return nil } raw, err := p.Load(sdkstep.DropKey(id)) if err != nil || raw == nil { return nil } var d nodebuildgo.Drop if json.Unmarshal(raw, &d) != nil { return nil } return &d } // refuseDropped ends a dropped ask the bus gave again: its step refused naming who dropped it, the message ended and the // drop's key deleted. func (p *Puller) refuseDropped(msg *Pulled, ask sdkstep.StepAsk, d *nodebuildgo.Drop) { key, err := sdkstep.AskStepKey(p.Node, ask.ID) if err == nil { at := now().UTC() step := sdkstep.Step{ID: ask.ID, Tool: ask.Kind + " ask", Node: p.Node, State: sdkstep.StepDropped, Started: at, Ended: at, Owner: buildrun.Owner, Run: int(msg.Delivered), Reason: (&droppedError{by: d.By, why: d.Why}).Error()} err = writeEnd(p.Put, key, step) } if err != nil { p.logf("the dropped ask %s cannot be refused on %s, so it is given back: %v", ask.ID, p.Node, err) _ = msg.Nak() return } p.logf("the ask %s is refused on %s: it was dropped (%s)", ask.ID, p.Node, d.Why) _ = msg.Term() p.forgetDrop(ask.ID) } // forgetDrop deletes an ask's drop key, said when it cannot be. func (p *Puller) forgetDrop(id string) { if p.ForgetPause == nil { return } if err := p.ForgetPause(sdkstep.DropKey(id)); err != nil { p.logf("the drop of the ask %s cannot be deleted: %v", id, err) } } // sweepDropped deletes, at the start, the drops older than DroppedKeep: their asks are gone from the stream. func (p *Puller) sweepDropped() { if p.PauseKeys == nil || p.ForgetPause == nil || p.Load == nil { return } keys, err := p.PauseKeys() if err != nil { p.logf("the pauses of %s cannot be listed to sweep the drops: %v", p.Node, err) return } for _, k := range keys { id, ok := strings.CutPrefix(k, sdkstep.DroppedPrefix) if !ok { continue } if d := p.droppedAsk(id); d != nil && (d.At == nil || now().Sub(*d.At) > DroppedKeep) { p.forgetDrop(id) } } }