mesh-delivery/modules/mesh-delivery-builder/buildrun/asks_test.go

package buildrun

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats.go"

	"git.novox.be/novox/mesh-delivery/sdk/buildstep"
	"git.novox.be/novox/mesh-sdk/go/busretry"
	"git.novox.be/novox/mesh-sdk/go/health"
)

func goAsk(id string) buildstep.StepAsk {
	return buildstep.StepAsk{ID: id, Kind: buildstep.AskDelivery, Language: buildstep.AskGo, Asker: "mesh-delivery-builder", At: time.Now(),
		Build: &buildstep.BuildJob{Module: "m", Path: ".", Fetch: &buildstep.Fetch{Repository: "novox/x", Commit: strings.Repeat("a", 40)},
			Artifacts: []buildstep.AskArtifact{{Artifact: "program"}}}}
}

// 1887/A4a: an asker puts the ask once and answers the step a build node wrote for it, naming the node; one that did not
// pass is an error naming the node and why; a wait of an ask already put puts nothing.
func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
	f := &FakeAsks{Answer: func(a buildstep.StepAsk) buildstep.Step {
		return buildstep.Step{ID: a.ID, Node: "ace", State: buildstep.StepPassed, Result: []byte(`{"artifacts":[]}`)}
	}}
	subjects := &subjectsOf{inner: f}
	k := Asker{Seats: testSeats(), Publisher: subjects, Waiter: f}
	var seen, also []buildstep.Step // issue 1890: a waiter that asks to see each value of the step sees it
	// Issue 1921: a second seen on the same context (a check's begin bound inside a try) is added, never in place of the first.
	both := WithStepSeen(WithStepSeen(context.Background(), func(s buildstep.Step) { seen = append(seen, s) }),
		func(s buildstep.Step) { also = append(also, s) })
	step, err := k.Ask(both, goAsk("mvl0frv18d1957"))
	if err != nil || step.Node != "ace" || len(f.Asks()) != 1 || len(seen) != 1 || seen[0].Node != "ace" || len(also) != 1 {
		t.Fatalf("step %+v, %v, asks %d, seen %+v and %+v", step, err, len(f.Asks()), seen, also)
	}
	// Design 47 G §481, §600: the ask is put under the builder's name, accept.<queue>.builder.
	if len(subjects.seen) != 1 || subjects.seen[0] != "mesh.seat.node-build-go.accept."+goAsk("x").Kind+".builder" {
		t.Fatalf("the ask was put on %v", subjects.seen)
	}
	f.Answer = func(a buildstep.StepAsk) buildstep.Step {
		return buildstep.Step{ID: a.ID, Node: "shanks", State: buildstep.StepRefused, Reason: "go 1.27 installed"}
	}
	if _, err := k.Ask(context.Background(), goAsk("mvl0frv18d1958")); err == nil || !strings.Contains(err.Error(), "shanks refused: go 1.27 installed") {
		t.Fatalf("a refused step: %v", err)
	}
	// Issue 1966 (ADR 0668 §1, §2; 1748 note 27): the put is a kept call under the ask's id, made again through a deadline and
	// no responders until the stream takes it; a refusal of the stream is not made again; a put that meets a transport fault
	// past the bound says it in buildstep.TransportFaultWords, so the key it fails is the hand tool's to build again.
	t.Cleanup(busretry.Waits(time.Second, time.Millisecond, time.Millisecond))
	for name, c := range map[string]struct {
		faults    []error
		tries     int
		transport bool
	}{
		"a deadline, then no responders, then taken": {faults: []error{context.DeadlineExceeded, nats.ErrNoResponders}, tries: 3},
		"refused by the stream":                      {faults: []error{errors.New("maximum messages exceeded")}, tries: 1},
		"a deadline past the bound":                  {faults: slices.Repeat([]error{nats.ErrTimeout}, 10000), transport: true},
	} {
		p := &faultyPuts{faults: c.faults, inner: &FakeAsks{}}
		restore := func() {}
		if c.transport {
			restore = busretry.Waits(20*time.Millisecond, time.Millisecond, time.Millisecond) // a bound this row passes
		}
		err := (Asker{Seats: testSeats(), Publisher: p, Waiter: p.inner}).Put(context.Background(), goAsk("mvl0frv18d1960"))
		restore()
		switch {
		case c.transport:
			if err == nil || !strings.HasPrefix(err.Error(), buildstep.TransportFaultWords) {
				t.Errorf("%s: %v, want it opens with the transport fault's words", name, err)
			}
		case c.tries == 1:
			if err == nil || p.tries != 1 || strings.Contains(err.Error(), buildstep.TransportFaultWords) {
				t.Errorf("%s: %v after %d tries, want one try and no transport fault", name, err, p.tries)
			}
		case err != nil || p.tries != c.tries || len(p.inner.Asks()) != 1 || len(p.ids) != c.tries || p.ids[0] != p.ids[len(p.ids)-1]:
			t.Errorf("%s: %v after %d tries under %v, %d asks taken, want %d tries under one id, one ask", name, err, p.tries, p.ids, len(p.inner.Asks()), c.tries)
		}
	}
	// Issue 1748 (ADR 0664 §3): a run that ended on a transport fault is no end; the next run's end is the answer, in either
	// arrival order; three runs then failed say a transport fault on 3 runs with each node; a bus that never gives the ask
	// again ends the wait saying so.
	for name, order := range map[string][]buildstep.Step{
		"again then passed": {{ID: "mvl0frv18d1971", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now()}, {ID: "mvl0frv18d1971", Node: "g14", State: buildstep.StepPassed, Run: 2}},
		"passed then again": {{ID: "mvl0frv18d1971", Node: "g14", State: buildstep.StepPassed, Run: 2}, {ID: "mvl0frv18d1971", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now()}},
	} {
		h := &FakeAsks{}
		for _, st := range order {
			h.Write(st)
		}
		if got, err := (Asker{Seats: testSeats(), Publisher: h, Waiter: h, Bound: time.Minute}).Wait(context.Background(), goAsk("mvl0frv18d1971")); err != nil || got.State != buildstep.StepPassed {
			t.Errorf("%s: %+v, %v", name, got, err)
		}
	}
	h := &FakeAsks{}
	for i, node := range []string{"ace", "g14", "shanks"} {
		st := buildstep.Step{ID: "mvl0frv18d1972", Node: node, State: buildstep.StepAgain, Run: i + 1, AgainAt: time.Now()}
		if i == 2 {
			st.State, st.Passing, st.Reason = buildstep.StepFailed, true, "lookup proxy: i/o timeout"
		}
		h.Write(st)
	}
	if _, err := (Asker{Seats: testSeats(), Publisher: h, Waiter: h, Bound: time.Minute}).Ask(context.Background(), goAsk("mvl0frv18d1972")); err == nil || !strings.Contains(err.Error(), "a transport fault on 3 runs (ace, g14, shanks): lookup proxy: i/o timeout") {
		t.Errorf("three runs then failed: %v", err)
	}
	h = &FakeAsks{}
	h.Write(buildstep.Step{ID: "mvl0frv18d1973", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now(), Started: time.Now()})
	if _, err := (Asker{Seats: testSeats(), Publisher: h, Waiter: h, Bound: 20 * time.Millisecond}).Wait(context.Background(), goAsk("mvl0frv18d1973")); err == nil || !strings.Contains(err.Error(), "given again after a transport fault") {
		t.Errorf("an ask never given again: %v", err)
	}
	// ADR 0664 §5: that ask is dropped; a wait ended by a named cause is dropped too, an ask no node took included (it waits in
	// its stream); a shutdown's plain cancel is not.
	drops := &seatDrops{}
	h = &FakeAsks{}
	h.Write(buildstep.Step{ID: "mvl0frv18d1974", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now(), Started: time.Now()})
	(Asker{Seats: testSeats(), Publisher: h, Waiter: h, Bound: 20 * time.Millisecond, Dropper: drops}).Wait(context.Background(), goAsk("mvl0frv18d1974"))
	named, stop := context.WithCancelCause(context.Background())
	h.Write(buildstep.Step{ID: "mvl0frv18d1975", Node: "g14", State: buildstep.StepRunning, Started: time.Now()})
	seenStep := WithStepSeen(named, func(buildstep.Step) { stop(Withdrawn{Why: "superseded by a newer fingerprint"}) })
	(Asker{Seats: testSeats(), Publisher: h, Waiter: h, Bound: time.Hour, Dropper: drops}).Wait(seenStep, goAsk("mvl0frv18d1975"))
	shut, off := context.WithCancel(context.Background())
	off()
	(Asker{Seats: testSeats(), Publisher: h, Waiter: h, Bound: time.Hour, Dropper: drops}).Wait(shut, goAsk("mvl0frv18d1977"))
	(Asker{Seats: testSeats(), Publisher: &FakeAsks{}, Waiter: &FakeAsks{}, Bound: time.Hour, Dropper: drops}).Wait(named, goAsk("mvl0frv18d1976"))
	// Review 1 (finding 1): a cause that is no withdrawal (the builder's check bound passing, which waits on) drops nothing.
	bounded, passed := context.WithCancelCause(context.Background())
	passed(errors.New("the check bound passed"))
	(Asker{Seats: testSeats(), Publisher: &FakeAsks{}, Waiter: &FakeAsks{}, Bound: time.Hour, Dropper: drops}).Wait(bounded, goAsk("mvl0frv18d1978"))
	if got := strings.Join(drops.got, "|"); !strings.Contains(got, "mvl0frv18d1974") || !strings.Contains(got, "mvl0frv18d1975 superseded by a newer fingerprint") || !strings.Contains(got, "mvl0frv18d1976 superseded by a newer fingerprint") || len(drops.got) != 3 {
		t.Errorf("drops %q, want the ask never given again, the superseded one and the one no node took, and no other", got)
	}
	// Review 1: a wait that starts after the first runs (a builder started again) reads the last value only, and its end still says
	// how many runs; a drop that failed is said, not discarded.
	late := &FakeAsks{}
	late.Write(buildstep.Step{ID: "mvl0frv18d1979", Node: "shanks", State: buildstep.StepFailed, Passing: true, Run: 3, Reason: "lookup proxy: i/o timeout"})
	if st, err := (Asker{Seats: testSeats(), Publisher: late, Waiter: late, Bound: time.Minute}).Wait(context.Background(), goAsk("mvl0frv18d1979")); err != nil || !strings.Contains(st.Reason, "a transport fault on 3 runs") {
		t.Errorf("a failed step of run 3 read alone says %q (%v), want it names 3 runs", st.Reason, err)
	}
	var said []string
	bad, off2 := context.WithCancelCause(context.Background())
	off2(Withdrawn{Why: "withdrawn"})
	(Asker{Seats: testSeats(), Publisher: &FakeAsks{}, Waiter: &FakeAsks{}, Bound: time.Hour, Dropper: failingDrops{}, Logf: func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }}).Wait(bad, goAsk("mvl0frv18d1980"))
	if len(said) != 1 || !strings.Contains(said[0], "mvl0frv18d1980") || !strings.Contains(said[0], "no holder") {
		t.Errorf("a drop that failed said %q, want it named with its cause", said)
	}
	// review round 2 item 1: an ask that waits in its stream for a free slot is not bounded; once a node writes its step
	// running, the step is bounded from its start, and the error names the node.
	g := &FakeAsks{}
	k = Asker{Seats: testSeats(), Publisher: g, Waiter: g, Bound: 20 * time.Millisecond}
	waiting, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer cancel()
	if _, err := k.Wait(waiting, goAsk("mvl0frv18d1959")); err != context.DeadlineExceeded {
		t.Fatalf("an ask no node took: %v, want it still waiting when ctx ended", err)
	}
	g.Write(buildstep.Step{ID: "mvl0frv18d1961", Node: "ace", State: buildstep.StepRunning, Started: time.Now()})
	if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1961")); err == nil || !strings.Contains(err.Error(), "on ace did not end within") {
		t.Fatalf("a step that never ends: %v", err)
	}
	if len(g.Asks()) != 0 {
		t.Fatal("a wait put an ask")
	}
	// Issue 1930: the bound is on a step that stands still, not on its age. A step past one bound from its start that entered a
	// new phase inside the bound is waited for until it ends; one that stood a whole bound in a phase fails at once, naming
	// the phase; one three bounds from its start ends, said as kept moving. The steps' own times stand for the clock, so the
	// bound is an hour no loaded machine reaches and no case waits on a pause (the check of 2026-11-02 22:35 tripped one
	// of 60 ms).
	k = Asker{Seats: testSeats(), Publisher: g, Waiter: g, Bound: time.Hour}
	now := time.Now()
	g.Write(buildstep.Step{ID: "mvl0frv18d1980", Node: "ace", State: buildstep.StepRunning, Started: now.Add(-90 * time.Minute), Phase: "pb", PhaseAt: now})
	g.Write(buildstep.Step{ID: "mvl0frv18d1980", Node: "ace", State: buildstep.StepPassed})
	if step, err := k.Wait(context.Background(), goAsk("mvl0frv18d1980")); err != nil || step.State != buildstep.StepPassed {
		t.Fatalf("a step that kept moving past its bound: %+v, %v", step, err)
	}
	g.Write(buildstep.Step{ID: "mvl0frv18d1981", Node: "ace", State: buildstep.StepRunning, Started: now.Add(-time.Hour), Phase: "integration: tests", PhaseAt: now.Add(-time.Hour)})
	if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1981")); err == nil || !strings.Contains(err.Error(), `stood still in "integration: tests"`) {
		t.Fatalf("a step that stood still in a phase: %v", err)
	}
	g.Write(buildstep.Step{ID: "mvl0frv18d1982", Node: "ace", State: buildstep.StepRunning, Started: now.Add(-3 * time.Hour), Phase: "qz", PhaseAt: now})
	if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1982")); err == nil || !strings.Contains(err.Error(), "kept moving") {
		t.Fatalf("a step that never stops moving: %v", err)
	}
	// Issue 734: an ask the build seat gave up on (DEAD_LETTERS keeps its copy with the ask's id) ends its wait as failed
	// with the words, though a node had taken it; under NotTakenBound one no node takes ends at the bound; a step written
	// first is answered as the node wrote it.
	dead := &DeadLetters{}
	lw := LostAskWaiter{Inner: g, Dead: dead}
	k = Asker{Seats: testSeats(), Publisher: g, Waiter: lw, Bound: time.Minute}
	g.Write(buildstep.Step{ID: "mvl0frv18d1970", Node: "ace", State: buildstep.StepRunning, Started: time.Now()})
	given, ok := GivenUpOf(map[string][]string{"Mesh-Dead-Msg-Id": {"mvl0frv18d1970"}, "Mesh-Dead-Deliveries": {"5"},
		"Mesh-Dead-Stream": {"SEAT_NODE_BUILD_GO"}, "Mesh-Dead-Gave-Up": {"2026-10-26T12:26:40Z"}})
	if !ok {
		t.Fatal("a dead letter's headers were not read")
	}
	go dead.Heard(given)
	step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1970"))
	if err != nil || step.State != buildstep.StepFailed || !IsLost(step) ||
		!strings.Contains(step.Reason, LostWords+" after 5 deliveries at 2026-10-26T12:26:40Z") {
		t.Fatalf("a given-up ask: %+v, %v", step, err)
	}
	if _, err := k.Passed(step, nil); err == nil || !strings.Contains(err.Error(), "on "+LostNode+" failed: "+LostWords) {
		t.Fatalf("a given-up ask's error: %v", err)
	}
	step, err = k.Wait(NotTakenBound(context.Background(), 20*time.Millisecond), goAsk("mvl0frv18d1971"))
	if err != nil || !IsNotTaken(step) || IsLost(step) {
		t.Fatalf("an ask no node took: %+v, %v", step, err)
	}
	g.Write(buildstep.Step{ID: "mvl0frv18d1972", Node: "g14", State: buildstep.StepPassed})
	if step, err = k.Wait(NotTakenBound(context.Background(), time.Minute), goAsk("mvl0frv18d1972")); err != nil || step.Node != "g14" {
		t.Fatalf("a step written: %+v, %v", step, err)
	}
	// Review round 1 of issue 734: a hearing the bus refuses at start is deaf and says so, never holds the start up, and
	// tries again on the next ask waited for; heard, it says so and hands the given-up asks on.
	fol := &refusingFollower{refuse: 1}
	var mu sync.Mutex
	var told []Hearing
	hearing := NewDeadLetterHearing(context.Background(), fol, []string{buildstep.AskGo}, dead, nil,
		func(h Hearing) { mu.Lock(); told = append(told, h); mu.Unlock() })
	hearing.Wait()
	if why := hearing.Deaf(); !strings.Contains(why, "SEAT_NODE_BUILD_GO: permissions violation") {
		t.Fatalf("a refused hearing says %q", why)
	}
	k = Asker{Seats: testSeats(), Publisher: g, Waiter: LostAskWaiter{Inner: g, Dead: dead, Hearing: hearing}, Bound: time.Minute}
	g.Write(buildstep.Step{ID: "mvl0frv18d1973", Node: "g14", State: buildstep.StepPassed})
	if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1973")); err != nil {
		t.Fatal(err)
	}
	hearing.Wait()
	mu.Lock()
	if hearing.Deaf() != "" || len(told) != 2 || told[0].Heard || !told[1].Heard || fol.calls() != 2 {
		t.Fatalf("after an ask: deaf %q, told %+v, followed %d times", hearing.Deaf(), told, fol.calls())
	}
	mu.Unlock()
	fol.heard(GivenUp{AskID: "mvl0frv18d1974", Deliveries: 5})
	step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1974"))
	if err != nil || !IsLost(step) {
		t.Fatalf("an ask heard given up after the hearing came back: %+v, %v", step, err)
	}
	// The lost check of 2026-10-28: a step whose own watch went deaf is read at the next nudge (a step of its build module
	// ended) and answered; a nudge of another build module reads nothing; a watch that closes is opened again.
	d := &FakeAsks{}
	heard := &HeardEnds{Inner: deafWatch{}, Read: d}
	k = Asker{Seats: testSeats(), Publisher: d, Waiter: heard, Bound: time.Minute}
	d.Write(buildstep.Step{ID: "mvl0frv18d1980", Node: "shanks", State: buildstep.StepPassed})
	got := make(chan buildstep.Step, 1)
	go func() { s, _ := k.Wait(context.Background(), goAsk("mvl0frv18d1980")); got <- s }()
	for heard.Waiting() == 0 {
		time.Sleep(time.Millisecond)
	}
	heard.Nudge(testNode)
	select {
	case s := <-got:
		t.Fatalf("a nudge of another build module answered %+v", s)
	case <-time.After(20 * time.Millisecond):
	}
	heard.Nudge(testGo)
	if s := <-got; s.State != buildstep.StepPassed || s.Node != "shanks" {
		t.Fatalf("a step its watch never handed over: %+v", s)
	}
	closing := &HeardEnds{Inner: &closingWatch{inner: d}, Read: d}
	if s, err := (Asker{Seats: testSeats(), Publisher: d, Waiter: closing}).Wait(context.Background(), goAsk("mvl0frv18d1980")); err != nil || s.State != buildstep.StepPassed {
		t.Fatalf("a watch that closed once: %+v, %v", s, err)
	}
	for _, row := range []struct {
		h    *Hearing
		want string
	}{{nil, health.StateUnhealthy}, {&told[0], health.StateUnhealthy}, {&told[1], health.StateHealthy}} {
		var raw []byte
		if row.h != nil {
			raw, _ = json.Marshal(row.h)
		}
		if a := HearingHealth(raw); a.State != row.want || a.Why == "" {
			t.Errorf("health of %+v = %+v, want %s", row.h, a, row.want)
		}
	}
}

// Review (issue 734 path): an ask taken again by the same node (its build module replaced mid-run) is bounded from the
// new run's start, not the first run's. The steps' own times stand for the clock: the first run started now, the second a
// whole bound ago, so the wait ends at once on the second run's bound where the first run's would wait an hour; no case
// waits on a pause a loaded machine can outlast (the check of 2026-11-02 22:35 tripped one of 60 ms). The context's
// bound is reached only by a wait that kept the first run's start.
func TestARedeliveryToTheSameNodeRestartsTheBound(t *testing.T) {
	f := &FakeAsks{}
	k := Asker{Seats: testSeats(), Publisher: f, Waiter: f, Bound: time.Hour}
	const id = "mvl0frv18d1962"
	now := time.Now()
	f.Write(buildstep.Step{ID: id, Node: "ace", State: buildstep.StepRunning, Started: now})
	f.Write(buildstep.Step{ID: id, Node: "ace", State: buildstep.StepRunning, Started: now.Add(-time.Hour)})
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	if _, err := k.Wait(ctx, goAsk(id)); err == nil || !strings.Contains(err.Error(), "on ace did not end within 1h0m0s of its start") {
		t.Fatalf("a redelivered run a bound old: %v, want its own bound to end the wait", err)
	}
}

// refusingFollower refuses the first `refuse` follows, as the bus does before the asker's grant is composed.
type refusingFollower struct {
	mu     sync.Mutex
	refuse int
	n      int
	hand   func(GivenUp)
}

func (f *refusingFollower) Follow(_ context.Context, _ string, _ time.Time, heard func(GivenUp)) (func(), error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.n++
	if f.n <= f.refuse {
		return nil, errors.New("permissions violation")
	}
	f.hand = heard
	return func() {}, nil
}

func (f *refusingFollower) calls() int { f.mu.Lock(); defer f.mu.Unlock(); return f.n }

func (f *refusingFollower) heard(g GivenUp) { f.mu.Lock(); h := f.hand; f.mu.Unlock(); h(g) }

// deafWatch is a watch of an ask's step that stays open and hears nothing.
type deafWatch struct{}

func (deafWatch) WatchAsk(context.Context, string, string) (<-chan buildstep.Step, error) {
	return make(chan buildstep.Step), nil
}

// closingWatch closes its first watch at once, and hands the later ones to inner.
type closingWatch struct {
	inner  AskWaiter
	closed bool
}

func (c *closingWatch) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
	if !c.closed {
		c.closed = true
		ch := make(chan buildstep.Step)
		close(ch)
		return ch, nil
	}
	return c.inner.WatchAsk(ctx, module, askID)
}

// seatDrops keeps the asks an Asker dropped: "<id> <why>".
type seatDrops struct{ got []string }

func (d *seatDrops) DropAsk(_ context.Context, a buildstep.StepAsk, why string) error {
	if strings.Contains(why, "bound") {
		why = ""
	}
	d.got = append(d.got, strings.TrimSpace(a.ID+" "+why))
	return nil
}

// failingDrops is an AskDropper whose verb is not answered.
type failingDrops struct{}

func (failingDrops) DropAsk(context.Context, buildstep.StepAsk, string) error {
	return errors.New("no holder answered")
}

// faultyPuts fails its first puts with faults, then puts through inner (issue 1966).
type faultyPuts struct {
	mu     sync.Mutex
	faults []error
	tries  int
	ids    []string
	inner  *FakeAsks
}

func (p *faultyPuts) PublishAsk(ctx context.Context, subject, id string, data []byte) error {
	p.mu.Lock()
	p.tries++
	p.ids = append(p.ids, id)
	var err error
	if len(p.faults) > 0 {
		err, p.faults = p.faults[0], p.faults[1:]
	}
	p.mu.Unlock()
	if err != nil {
		return err
	}
	return p.inner.PublishAsk(ctx, subject, id, data)
}

// subjectsOf records the subject each ask is published on.
type subjectsOf struct {
	inner AskPublisher
	seen  []string
}

func (p *subjectsOf) PublishAsk(ctx context.Context, subject, id string, data []byte) error {
	p.seen = append(p.seen, subject)
	return p.inner.PublishAsk(ctx, subject, id, data)
}