mesh-catalog/modules/mesh-build-go/buildstep/pull_drop.go

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.<ask id> (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 {
		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)
		}
	}
}