C1b, C1c: the seat's verb drop refuses a dropped ask at pull, ends one that runs, and the start sweeps old drops (ADR 0664 §5, issue 1748)

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

modules/mesh-build-go/buildstep/pull.go

@@ -77,6 +77,10 @@ type Puller struct {
 	// PausePut writes the node's pause key: the module's pauses state (sdkstep.PausesState), which has no age, so a
 	// pause never ends by itself while the steps state drops its old keys. Put when nil.
 	PausePut Put
+	// ForgetPause deletes one key of the pauses state and PauseKeys lists its keys (nil for none): a dropped ask's key is
+	// deleted when the ask is refused at pull, and the start sweeps the keys of asks the stream no longer holds (Drop).
+	ForgetPause func(key string) error
+	PauseKeys   func() ([]string, error)
 	// WatchConfig, when set, follows the module's config value `capacity` (CapacitySetting) as mesh-config resolves it
 	// for this node, and wakes the puller on each change (watchCapacity): WatchConfigServed in a module, a test's own
 	// double. Until it hands a value the puller judges by Capacity.
@@ -112,6 +116,8 @@ type Puller struct {
 	mu      sync.Mutex
 	inHand  int
 	running map[string]RunningAsk
+	// cancels end a running ask's step by its id: the seat's verb drop (Drop).
+	cancels map[string]context.CancelCauseFunc
 	paused  bool
 	wake    chan struct{}
 	once    sync.Once
@@ -428,6 +434,7 @@ func (p *Puller) Pull(ctx context.Context) error {
 		// node kept from building is held by its capacity (0), not by a pause.
 		p.logf("the pause of %s cannot be read: %v; pulling as not paused", p.Node, err)
 	}
+	p.sweepDropped()
 	p.follow() // WatchConfigServed waits for Serve, which a module starts after its pull
 	go p.watchSettings(ctx)
 	wait := p.PullWait
@@ -615,6 +622,10 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
 		p.refuseUnread(msg, err)
 		return
 	}
+	if d := p.droppedAsk(ask.ID); d != nil {
+		p.refuseDropped(msg, ask, d)
+		return
+	}
 	if lacks := sdkstep.Lacks(ask.Needs, p.Has); len(lacks) > 0 {
 		p.decline(msg, ask, lacks)
 		return
@@ -677,11 +688,20 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key
 		}
 	}()
 	ph := &phases{put: p.Put, key: key, step: step, logf: p.Logf, prefix: fmt.Sprintf("the %s ask %s on %s", ask.Kind, ask.ID, p.Node)}
-	result, err := protect(ctx, func(ctx context.Context) (any, error) {
+	runCtx := p.cancellable(ctx, ask.ID)
+	result, err := protect(runCtx, func(ctx context.Context) (any, error) {
 		return p.Run(context.WithValue(ctx, phaseKey{}, ph), ask)
 	})
 	step = ph.end() // its end keeps the phase it ended in
 	close(done)
+	if d := droppedCause(runCtx); d != nil && ctx.Err() == nil {
+		// The seat's verb drop ended it (ADR 0664 §5): refused, naming who dropped it and why; acknowledged, never given again.
+		err = &Refusal{Text: d.Error()}
+		defer p.forgetDrop(ask.ID)
+	}
+	p.mu.Lock()
+	delete(p.cancels, ask.ID)
+	p.mu.Unlock()
 	var dec *Declining
 	if errors.As(err, &dec) && ctx.Err() == nil {
 		p.declineRun(msg, ask, key, step, dec)

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

@@ -0,0 +1,147 @@
+package buildstep
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	buildrun "git.novox.be/novox/mesh-delivery/sdk/buildrun"
+	sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
+)
+
+// **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 refused naming who dropped it and why.
+
+// 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")
+	}
+	raw, err := json.Marshal(sdkstep.Drop{V: sdkstep.Version, By: by, Why: why, At: now().UTC()})
+	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
+}
+
+// droppedAsk is the drop kept for the ask, nil for none (or no state to read it from).
+func (p *Puller) droppedAsk(id string) *sdkstep.Drop {
+	if p.Load == nil {
+		return nil
+	}
+	raw, err := p.Load(sdkstep.DropKey(id))
+	if err != nil || raw == nil {
+		return nil
+	}
+	var d sdkstep.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 *sdkstep.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.StepRefused, 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 && now().Sub(d.At) > DroppedKeep {
+			p.forgetDrop(id)
+		}
+	}
+}

modules/mesh-build-go/buildstep/pull_seat_test.go

@@ -2,6 +2,10 @@ package buildstep
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -38,3 +42,93 @@ func TestSeatAsksPullsTheKindsWorkerAndSettlesByID(t *testing.T) {
 		t.Errorf("settled %v", settled)
 	}
 }
+
+// pauseKV is a pauses state in memory: its put, load, delete and keys.
+type pauseKV struct {
+	mu sync.Mutex
+	m  map[string][]byte
+}
+
+func (k *pauseKV) put(key string, v any) (uint64, error) {
+	raw, _ := v.(json.RawMessage)
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	if k.m == nil {
+		k.m = map[string][]byte{}
+	}
+	k.m[key] = raw
+	return 1, nil
+}
+func (k *pauseKV) load(key string) ([]byte, error) {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	return k.m[key], nil
+}
+func (k *pauseKV) forget(key string) error {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	delete(k.m, key)
+	return nil
+}
+func (k *pauseKV) keys() ([]string, error) {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	var out []string
+	for key := range k.m {
+		out = append(out, key)
+	}
+	return out, nil
+}
+
+// ADR 0664 §5 (issue 1748): the seat's verb drop. An ask dropped before it is given is refused at pull, naming who dropped it
+// and why, its key deleted, and never run; one that runs when it is dropped has its step's context cancelled and ends
+// refused the same way, acknowledged; the start deletes drops older than DroppedKeep.
+func TestADroppedAskIsRefusedAtPullAndARunningOneIsEnded(t *testing.T) {
+	for _, running := range []bool{false, true} {
+		src, st, kv := NewFakeAsks(), newSteps(), &pauseKV{}
+		a := testAsk(sdkstep.AskDelivery)
+		started := make(chan struct{})
+		ran := false
+		p := &Puller{Source: src, Language: sdkstep.AskGo, Node: "shanks", Put: st.put, PausePut: kv.put, Load: kv.load, ForgetPause: kv.forget, PauseKeys: kv.keys,
+			Capacity: func() int { return 1 }, PullWait: 10 * time.Millisecond,
+			Run: func(ctx context.Context, _ sdkstep.StepAsk) (any, error) {
+				ran = true
+				close(started)
+				<-ctx.Done()
+				return nil, ctx.Err()
+			}}
+		if !running {
+			if err := p.Drop(a.ID, "mesh-delivery-builder", "superseded"); err != nil {
+				t.Fatal(err)
+			}
+		}
+		_ = src.PublishAsk(a)
+		ctx, cancel := context.WithCancel(context.Background())
+		go p.Pull(ctx)
+		if running {
+			<-started
+			if err := p.Drop(a.ID, "mesh-delivery-builder", "superseded"); err != nil {
+				t.Fatal(err)
+			}
+		}
+		waitFor(t, "the ask ended", func() bool { ack, tm, _ := src.Fates(); return len(ack)+len(tm) == 1 })
+		cancel()
+		key, _ := sdkstep.AskStepKey("shanks", a.ID)
+		s, ok := st.get(key)
+		if !ok || s.State != sdkstep.StepRefused || !strings.Contains(s.Reason, "dropped by mesh-delivery-builder: superseded") || ran != running {
+			t.Errorf("running %v: step %+v (%v), ran %v", running, s, ok, ran)
+		}
+		waitFor(t, fmt.Sprint("the drop key deleted, running ", running), func() bool { kv.mu.Lock(); defer kv.mu.Unlock(); _, held := kv.m[sdkstep.DropKey(a.ID)]; return !held })
+	}
+	// The start deletes a drop older than DroppedKeep and keeps a newer one.
+	kv := &pauseKV{}
+	old, _ := json.Marshal(sdkstep.Drop{V: sdkstep.Version, Why: "old", At: time.Now().Add(-DroppedKeep - time.Hour)})
+	fresh, _ := json.Marshal(sdkstep.Drop{V: sdkstep.Version, Why: "fresh", At: time.Now()})
+	kv.put(sdkstep.DropKey("old"), json.RawMessage(old))
+	kv.put(sdkstep.DropKey("fresh"), json.RawMessage(fresh))
+	p := &Puller{Load: kv.load, ForgetPause: kv.forget, PauseKeys: kv.keys}
+	p.sweepDropped()
+	if keys, _ := kv.keys(); len(keys) != 1 || keys[0] != sdkstep.DropKey("fresh") {
+		t.Errorf("keys after the sweep = %v, want the fresh drop alone", keys)
+	}
+}

modules/mesh-build-go/buildstep/seat_tools.go

@@ -18,7 +18,20 @@ func SeatTools(seat, what string, p *Puller) []stdio.Tool {
 		return map[string]any{"node": p.Node, "capacity": c.Effective, "capacity_why": c.Why, "capacity_detail": c,
 			"running": p.Running(), "paused": p.Paused()}, nil
 	}
-	return []stdio.Tool{
+	drop := stdio.Tool{Name: seat + ".drop", Description: "End an " + what + " ask its caller no longer waits for: it is kept as dropped on this node " +
+		"(so a bus that gives it again has it refused), and a step running here for it is ended, written refused naming who dropped it and why.",
+		Input: map[string]any{"type": "object", "required": []string{"id", "why"}, "properties": map[string]any{
+			"id":  map[string]any{"type": "string", "description": "the ask's id"},
+			"by":  map[string]any{"type": "string", "description": "who drops it"},
+			"why": map[string]any{"type": "string", "description": "why the caller no longer waits"}}},
+		Run: func(args map[string]any) (any, error) {
+			str := func(k string) string { s, _ := args[k].(string); return s }
+			if err := p.Drop(str("id"), str("by"), str("why")); err != nil {
+				return nil, err
+			}
+			return map[string]any{}, nil
+		}}
+	return []stdio.Tool{drop,
 		verb("current", "The "+what+" asks this node runs now (each one's id, kind and when it started), its build capacity (the configured one, "+
 			"bounded by the cores its node's build slice gives over the cores a step needs) and why, "+
 			"and whether it is paused.", current),

modules/mesh-build-go/cmd/mesh-build-go/asks.go

@@ -113,7 +113,7 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
 			return raw, err
 		}
 	}
-	return &buildstep.Puller{Source: source, Language: sdkstep.AskGo, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load,
+	return &buildstep.Puller{Source: source, Language: sdkstep.AskGo, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load, ForgetPause: m.ForgetPause, PauseKeys: m.ReadPauses.Keys,
 		Capacity: capacity, SettingsFile: settings, WatchConfig: watch, CoresPerStep: cores, BuildHours: hours, Logf: m.Say, Has: m.Has,
 		Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
 			if err := m.Check(ctx); err != nil {

modules/mesh-build-go/cmd/mesh-build-go/main.go

@@ -44,6 +44,7 @@ func main() {
 		ReadSteps:           readOf(stdio.State(seat.StateSteps)),
 		Pauses:              stdio.State(sdkstep.PausesState).Put,
 		ReadPauses:          readOf(stdio.State(sdkstep.PausesState)),
+		ForgetPause:         stdio.State(sdkstep.PausesState).Delete,
 		ReadToolchainsState: readOf(stdio.State(seat.StateToolchains)),
 		Ctx:                 ctx,
 		Say:                 logf,

modules/mesh-build-go/cmd/mesh-build-go/module.go

@@ -35,6 +35,8 @@ type Module struct {
 	// Pauses is the Put of the pauses state, which keeps the node's pause key without an age; ReadPauses reads it back.
 	Pauses     buildstep.Put
 	ReadPauses StateRead
+	// ForgetPause deletes one key of the pauses state: a dropped ask's, once it is refused (the seat's verb drop).
+	ForgetPause func(key string) error
 	// Emit tells the bus an event (stdio.Emit); nil emits nothing. The guard emits toolchain.reported after each write of
 	// the toolchains key, so the indexer, which holds no timer, learns that the bucket exists (issue 1444).
 	Emit func(eventType string, body any) error

modules/mesh-build-image/cmd/mesh-build-image/asks.go

@@ -62,7 +62,7 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
 	if m.ReadPauses.Get != nil {
 		load = func(key string) ([]byte, error) { return m.ReadPauses.Get(key) }
 	}
-	return &buildstep.Puller{Source: source, Language: sdkstep.AskImage, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load,
+	return &buildstep.Puller{Source: source, Language: sdkstep.AskImage, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load, ForgetPause: m.ForgetPause, PauseKeys: m.ReadPauses.Keys,
 		Capacity: capacity, SettingsFile: settings, WatchConfig: watch, CoresPerStep: cores, BuildHours: hours, Eligible: m.Eligible, Logf: m.Say,
 		Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
 			if err := m.Check(ctx); err != nil {

modules/mesh-build-image/cmd/mesh-build-image/main.go

@@ -32,15 +32,16 @@ func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, "[mesh-build-image]
 func main() {
 	ctx := context.Background()
 	m := &Module{
-		Node:       strings.TrimSpace(os.Getenv("MESH_NODE")),
-		Steps:      stdio.State(seat.StateSteps).Put,
-		ReadSteps:  readOf(stdio.State(seat.StateSteps)),
-		Pauses:     stdio.State(sdkstep.PausesState).Put,
-		ReadPauses: readOf(stdio.State(sdkstep.PausesState)),
-		Ctx:        ctx,
-		Say:        logf,
-		Work:       WorkDir,
-		Store:      StoreDir,
+		Node:        strings.TrimSpace(os.Getenv("MESH_NODE")),
+		Steps:       stdio.State(seat.StateSteps).Put,
+		ReadSteps:   readOf(stdio.State(seat.StateSteps)),
+		Pauses:      stdio.State(sdkstep.PausesState).Put,
+		ReadPauses:  readOf(stdio.State(sdkstep.PausesState)),
+		ForgetPause: stdio.State(sdkstep.PausesState).Delete,
+		Ctx:         ctx,
+		Say:         logf,
+		Work:        WorkDir,
+		Store:       StoreDir,
 	}
 	m.Floor = floorOf(m.Node)
 	if u, err := user.Current(); err == nil {

modules/mesh-build-image/cmd/mesh-build-image/module.go

@@ -30,6 +30,8 @@ type Module struct {
 	// Pauses is the Put of the pauses state, which keeps the node's pause key without an age; ReadPauses reads it back.
 	Pauses     buildstep.Put
 	ReadPauses StateRead
+	// ForgetPause deletes one key of the pauses state: a dropped ask's, once it is refused (the seat's verb drop).
+	ForgetPause func(key string) error
 	// Ctx is the process's context, which a step runs under (the tool call's is cut at 30 s, the step's is not).
 	Ctx context.Context
 	Say func(format string, a ...any)

modules/mesh-build-node/cmd/mesh-build-node/asks.go

@@ -72,7 +72,7 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
 	if m.ReadPauses.Get != nil {
 		load = func(key string) ([]byte, error) { return m.ReadPauses.Get(key) }
 	}
-	return &buildstep.Puller{Source: source, Language: sdkstep.AskNode, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load,
+	return &buildstep.Puller{Source: source, Language: sdkstep.AskNode, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load, ForgetPause: m.ForgetPause, PauseKeys: m.ReadPauses.Keys,
 		Capacity: capacity, SettingsFile: settings, WatchConfig: watch, CoresPerStep: cores, BuildHours: hours, Logf: m.Say,
 		Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
 			if err := m.Prepare(ctx); err != nil {

modules/mesh-build-node/cmd/mesh-build-node/main.go

@@ -50,6 +50,7 @@ func main() {
 		ReadSteps:           readOf(stdio.State(seat.StateSteps)),
 		Pauses:              stdio.State(sdkstep.PausesState).Put,
 		ReadPauses:          readOf(stdio.State(sdkstep.PausesState)),
+		ForgetPause:         stdio.State(sdkstep.PausesState).Delete,
 		ReadToolchainsState: readOf(stdio.State(seat.StateToolchains)),
 		Work:                WorkDir,
 		Run:                 dforgekey.ExecRunnerClean(0),

modules/mesh-build-node/cmd/mesh-build-node/module.go

@@ -49,6 +49,8 @@ type Module struct {
 	// Pauses is the Put of the pauses state, which keeps the node's pause key without an age; ReadPauses reads it back.
 	Pauses     buildstep.Put
 	ReadPauses StateRead
+	// ForgetPause deletes one key of the pauses state: a dropped ask's, once it is refused (the seat's verb drop).
+	ForgetPause func(key string) error
 	// Emit tells the bus an event (stdio.Emit); nil emits nothing. The guard emits toolchain.reported after each write of
 	// the toolchains key, so the indexer, which holds no timer, learns that the bucket exists (issue 1444).
 	Emit func(eventType string, body any) error