fix(hints): discard detector processes interrupted at response boundaries

commit: 8f581cc7536694696af69f34a09eb28c452c99aa
author: brice · 15h ago · 2026-10-22 20:28 UTC
parent: 1ade7b5
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +92 -9

M internal/adapters/hints/detector.go +30 -9


@@ -283,6 +283,24 @@ func (d procDetector) cmd() exec.Cmd {
283	283		return exec.Command(d.command, args...)
284	284		}
285	285
	286	+	// interruptOnDeadline binds both callbacks to this process only. Stopping joins
	287	+	// any callback that already started, so its result can never be accepted.
	288	+	func interruptOnDeadline(ctx context.Context, timeout time.Duration, p *process) func() (timedOut, canceled bool) {
	289	+	timerDone, contextDone := make(chan struct{}), make(chan struct{})
	290	+	timer := time.AfterFunc(timeout, func() { defer close(timerDone); p.interrupt() })
	291	+	stop := context.AfterFunc(ctx, func() { defer close(contextDone); p.interrupt() })
	292	+	return func() (bool, bool) {
	293	+	timedOut, canceled := !timer.Stop(), !stop()
	294	+	if timedOut {
	295	+	<-timerDone
	296	+	}
	297	+	if canceled {
	298	+	<-contextDone
	299	+	}
	300	+	return timedOut, canceled
	301	+	}
	302	+	}
	303	+
286	304		func (d *procDetector) start(ctx context.Context) error {
287	305		log := logging.For(ctx, "hints")
288	306		cmd := d.cmd()

@@ -311,12 +329,10 @@ func (d *procDetector) start(ctx context.Context) error {
311	329		}
312	330		p := &process{cmd: cmd, in: inW, out: outR, rd: bufio.NewReaderSize(outR, 1<<16)}
313	331		t0 := time.Now()
314		-	timer := time.AfterFunc(d.helloTimeout, p.interrupt)
315		-	stop := context.AfterFunc(ctx, p.interrupt)
	332	+	stop := interruptOnDeadline(ctx, d.helloTimeout, p)
316	333		device, err := readHello(p.rd)
317		-	stop()
318		-	timedOut := !timer.Stop()
319		-	if err != nil {
	334	+	timedOut, canceled := stop()
	335	+	if err != nil \|\| timedOut \|\| canceled {
320	336		p.kill()
321	337		if ctx.Err() != nil {
322	338		err = ctx.Err()

@@ -338,11 +354,16 @@ func (d *procDetector) Detect(ctx context.Context, f detFrame) ([]detection, err
338	354		}
339	355		p := d.proc
340	356		d.seq++
341		-	timer := time.AfterFunc(d.reqTimeout, p.interrupt)
342		-	stop := context.AfterFunc(ctx, p.interrupt)
	357	+	stop := interruptOnDeadline(ctx, d.reqTimeout, p)
343	358		dets, err := d.roundTrip(p, f)
344		-	stop()
345		-	timedOut := !timer.Stop()
	359	+	timedOut, canceled := stop()
	360	+	if timedOut \|\| canceled {
	361	+	d.drop()
	362	+	if ctx.Err() != nil {
	363	+	return nil, ctx.Err()
	364	+	}
	365	+	return nil, fmt.Errorf("hints: detector timed out after %s", d.reqTimeout)
	366	+	}
346	367		if err == nil {
347	368		return dets, nil
348	369		}

 	return exec.Command(d.command, args...)
 }
+// interruptOnDeadline binds both callbacks to this process only. Stopping joins
+// any callback that already started, so its result can never be accepted.
+func interruptOnDeadline(ctx context.Context, timeout time.Duration, p *process) func() (timedOut, canceled bool) {
+	timerDone, contextDone := make(chan struct{}), make(chan struct{})
+	timer := time.AfterFunc(timeout, func() { defer close(timerDone); p.interrupt() })
+	stop := context.AfterFunc(ctx, func() { defer close(contextDone); p.interrupt() })
+	return func() (bool, bool) {
+		timedOut, canceled := !timer.Stop(), !stop()
+		if timedOut {
+			<-timerDone
+		}
+		if canceled {
+			<-contextDone
+		}
+		return timedOut, canceled
+	}
+}
 func (d *procDetector) start(ctx context.Context) error {
 	log := logging.For(ctx, "hints")
 	cmd := d.cmd()
 	}
 	p := &process{cmd: cmd, in: inW, out: outR, rd: bufio.NewReaderSize(outR, 1<<16)}
 	t0 := time.Now()
-	timer := time.AfterFunc(d.helloTimeout, p.interrupt)
+	stop := interruptOnDeadline(ctx, d.helloTimeout, p)
-	stop := context.AfterFunc(ctx, p.interrupt)
 	device, err := readHello(p.rd)
-	stop()
+	timedOut, canceled := stop()
-	timedOut := !timer.Stop()
+	if err != nil || timedOut || canceled {
-	if err != nil {
 		p.kill()
 		if ctx.Err() != nil {
 			err = ctx.Err()
 	}
 	p := d.proc
 	d.seq++
-	timer := time.AfterFunc(d.reqTimeout, p.interrupt)
+	stop := interruptOnDeadline(ctx, d.reqTimeout, p)
-	stop := context.AfterFunc(ctx, p.interrupt)
 	dets, err := d.roundTrip(p, f)
-	stop()
+	timedOut, canceled := stop()
-	timedOut := !timer.Stop()
+	if timedOut || canceled {
+		d.drop()
+		if ctx.Err() != nil {
+			return nil, ctx.Err()
+		}
+		return nil, fmt.Errorf("hints: detector timed out after %s", d.reqTimeout)
+	}
 	if err == nil {
 		return dets, nil
 	}

M internal/adapters/hints/detector_test.go +62 -0


@@ -11,6 +11,7 @@ import (
11	11		"os"
12	12		"path/filepath"
13	13		"testing"
	14	+	"testing/synctest"
14	15		"time"
15	16
16	17		"github.com/stretchr/testify/require"

@@ -271,3 +272,64 @@ func TestDetectorCloseClosesOutput(t *testing.T) {
271	272		require.ErrorIs(t, err, os.ErrClosed)
272	273		require.NoError(t, d.Close())
273	274		}
	275	+
	276	+	// A complete buffered response must not override an already launched context
	277	+	// callback, including a detector-reported error that normally keeps the child.
	278	+	func TestDetectorCanceledBufferedResponse(t *testing.T) {
	279	+	for _, status := range []uint32{0, 1} {
	280	+	t.Run(fmt.Sprint(status), func(t *testing.T) {
	281	+	d, l := childDetector(t, "close")
	282	+	require.NoError(t, d.start(context.Background()))
	283	+	conn, err := l.Accept()
	284	+	require.NoError(t, err)
	285	+	defer conn.Close()
	286	+	p := d.proc
	287	+	response := cat([]byte("NHRS"), u32(1), u64(1), u32(status))
	288	+	if status == 0 {
	289	+	response = cat(response, f32(0), u32(0))
	290	+	} else {
	291	+	response = cat(response, u32(3), []byte("bad"))
	292	+	}
	293	+	p.rd.Reset(bytes.NewReader(response))
	294	+	ctx, cancel := context.WithCancel(context.Background())
	295	+	cancel()
	296	+	_, err = d.Detect(ctx, detFrame{width: 1, height: 1, stride: 4, m: &mapping{b: make([]byte, 4)}})
	297	+	require.ErrorIs(t, err, context.Canceled)
	298	+	require.Nil(t, d.proc)
	299	+	require.True(t, p.reaped)
	300	+	})
	301	+	}
	302	+	}
	303	+
	304	+	func TestInterruptDeadlineBoundary(t *testing.T) {
	305	+	for _, cancelContext := range []bool{false, true} {
	306	+	t.Run(fmt.Sprint(cancelContext), func(t *testing.T) {
	307	+	synctest.Test(t, func(t *testing.T) {
	308	+	inR, inW, err := os.Pipe()
	309	+	require.NoError(t, err)
	310	+	defer inR.Close()
	311	+	outR, outW, err := os.Pipe()
	312	+	require.NoError(t, err)
	313	+	defer outW.Close()
	314	+	// Already reaped: only the real pipe endpoints remain to interrupt.
	315	+	p := &process{in: inW, out: outR, reaped: true}
	316	+	ctx, cancel := context.WithCancel(context.Background())
	317	+	defer cancel()
	318	+	stop := interruptOnDeadline(ctx, time.Second, p)
	319	+	if cancelContext {
	320	+	cancel()
	321	+	} else {
	322	+	time.Sleep(time.Second)
	323	+	}
	324	+	synctest.Wait()
	325	+	timedOut, canceled := stop()
	326	+	require.Equal(t, !cancelContext, timedOut)
	327	+	require.Equal(t, cancelContext, canceled)
	328	+	_, err = inW.Write([]byte{1})
	329	+	require.ErrorIs(t, err, os.ErrClosed)
	330	+	_, err = outR.Read(make([]byte, 1))
	331	+	require.ErrorIs(t, err, os.ErrClosed)
	332	+	})
	333	+	})
	334	+	}
	335	+	}


@@ -11,6 +11,7 @@ import (
11	"os"	11		"os"
12	"path/filepath"	12		"path/filepath"
13	"testing"	13		"testing"
		14	+	"testing/synctest"
14	"time"	15		"time"
15		16
16	"github.com/stretchr/testify/require"	17		"github.com/stretchr/testify/require"

@@ -271,3 +272,64 @@ func TestDetectorCloseClosesOutput(t *testing.T) {
271	require.ErrorIs(t, err, os.ErrClosed)	272		require.ErrorIs(t, err, os.ErrClosed)
272	require.NoError(t, d.Close())	273		require.NoError(t, d.Close())
273	}	274		}
		275	+
		276	+	// A complete buffered response must not override an already launched context
		277	+	// callback, including a detector-reported error that normally keeps the child.
		278	+	func TestDetectorCanceledBufferedResponse(t *testing.T) {
		279	+	for _, status := range []uint32{0, 1} {
		280	+	t.Run(fmt.Sprint(status), func(t *testing.T) {
		281	+	d, l := childDetector(t, "close")
		282	+	require.NoError(t, d.start(context.Background()))
		283	+	conn, err := l.Accept()
		284	+	require.NoError(t, err)
		285	+	defer conn.Close()
		286	+	p := d.proc
		287	+	response := cat([]byte("NHRS"), u32(1), u64(1), u32(status))
		288	+	if status == 0 {
		289	+	response = cat(response, f32(0), u32(0))
		290	+	} else {
		291	+	response = cat(response, u32(3), []byte("bad"))
		292	+	}
		293	+	p.rd.Reset(bytes.NewReader(response))
		294	+	ctx, cancel := context.WithCancel(context.Background())
		295	+	cancel()
		296	+	_, err = d.Detect(ctx, detFrame{width: 1, height: 1, stride: 4, m: &mapping{b: make([]byte, 4)}})
		297	+	require.ErrorIs(t, err, context.Canceled)
		298	+	require.Nil(t, d.proc)
		299	+	require.True(t, p.reaped)
		300	+	})
		301	+	}
		302	+	}
		303	+
		304	+	func TestInterruptDeadlineBoundary(t *testing.T) {
		305	+	for _, cancelContext := range []bool{false, true} {
		306	+	t.Run(fmt.Sprint(cancelContext), func(t *testing.T) {
		307	+	synctest.Test(t, func(t *testing.T) {
		308	+	inR, inW, err := os.Pipe()
		309	+	require.NoError(t, err)
		310	+	defer inR.Close()
		311	+	outR, outW, err := os.Pipe()
		312	+	require.NoError(t, err)
		313	+	defer outW.Close()
		314	+	// Already reaped: only the real pipe endpoints remain to interrupt.
		315	+	p := &process{in: inW, out: outR, reaped: true}
		316	+	ctx, cancel := context.WithCancel(context.Background())
		317	+	defer cancel()
		318	+	stop := interruptOnDeadline(ctx, time.Second, p)
		319	+	if cancelContext {
		320	+	cancel()
		321	+	} else {
		322	+	time.Sleep(time.Second)
		323	+	}
		324	+	synctest.Wait()
		325	+	timedOut, canceled := stop()
		326	+	require.Equal(t, !cancelContext, timedOut)
		327	+	require.Equal(t, cancelContext, canceled)
		328	+	_, err = inW.Write([]byte{1})
		329	+	require.ErrorIs(t, err, os.ErrClosed)
		330	+	_, err = outR.Read(make([]byte, 1))
		331	+	require.ErrorIs(t, err, os.ErrClosed)
		332	+	})
		333	+	})
		334	+	}
		335	+	}