Supervise quarantined Remote input helpers and block uncertain replacement

internal/platform/xorg/remote_helper_native_test.go

@@ -6,6 +6,8 @@ package xorg
 
 import (
 	"bytes"
+	"context"
+	"errors"
 	"io"
 	"os"
 	"os/exec"
@@ -16,6 +18,68 @@ import (
 	"git.stormux.org/storm/cthulhu/internal/remoteinput"
 )
 
+func TestNativeRemoteKeyHelperSupervisor(t *testing.T) {
+	for _, cancellation := range []string{"explicit", "lease-context"} {
+		t.Run(cancellation, func(t *testing.T) {
+			remoteNativeThread(t)
+			path := os.Getenv("CTHULHU_REMOTE_INPUT_HELPER")
+			supervisor, err := remoteinput.NewSupervisor(path)
+			if err != nil {
+				t.Fatal(err)
+			}
+			probe := remoteNativeTarget(t)
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			client, err := supervisor.Start(ctx, 71)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer func() { client.Cancel(); client.Wait() }()
+			codes := make([]uint16, 0, 2)
+			for _, name := range []string{"Shift_L", "Control_R"} {
+				code, err := probe.backend.Resolve(name)
+				if err != nil {
+					t.Fatal(err)
+				}
+				codes = append(codes, code)
+				if err := client.Apply(t.Context(), name, true); err != nil {
+					t.Fatal(err)
+				}
+				remoteNativeHeld(t, probe, code, true)
+			}
+			if _, err := supervisor.Start(t.Context(), 72); !errors.Is(err, remoteinput.ErrUnavailable) {
+				t.Fatal("active native helper replaced", err)
+			}
+			if cancellation == "explicit" {
+				client.Cancel()
+			} else {
+				cancel()
+			}
+			if err := client.Wait(); err != nil {
+				t.Fatal("native cleanup not confirmed", err)
+			}
+			for _, code := range codes {
+				remoteNativeHeld(t, probe, code, false)
+			}
+			replacement, err := supervisor.Start(t.Context(), 72)
+			if err != nil {
+				t.Fatal("confirmed cleanup blocked fresh owner", err)
+			}
+			defer func() { replacement.Cancel(); replacement.Wait() }()
+			client.Cancel()
+			if err := replacement.Apply(t.Context(), "Shift_L", true); err != nil {
+				t.Fatal("old owner cancellation affected replacement", err)
+			}
+			remoteNativeHeld(t, probe, codes[0], true)
+			replacement.Cancel()
+			if err := replacement.Wait(); err != nil {
+				t.Fatal(err)
+			}
+			remoteNativeHeld(t, probe, codes[0], false)
+		})
+	}
+}
+
 // The writer owns both write ends exclusively. The supervising test owns the
 // helper process so it can observe cleanup and reap it after writer SIGKILL.
 func TestNativeRemoteKeyControlWriter(t *testing.T) {

internal/remoteinput/supervisor.go

@@ -0,0 +1,365 @@
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package remoteinput
+
+import (
+	"context"
+	"errors"
+	"io"
+	"math"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"sync"
+	"syscall"
+	"time"
+)
+
+var (
+	ErrUnavailable = errors.New("remote input helper unavailable")
+	ErrUncertain   = errors.New("remote input helper retirement unconfirmed")
+)
+
+const operationTimeout = time.Second
+const retirementTimeout = Watchdog + time.Second
+
+// Supervisor is a quarantined reader-side process owner. It grants no consent
+// and is not attached to the production key mailbox. A future owner must bind
+// ctx/epoch to an approved lease and recheck native policy before every Apply.
+// Unconfirmed cleanup permanently blocks replacement on this Supervisor.
+type Supervisor struct {
+	mu        sync.Mutex
+	active    bool
+	uncertain bool
+	command   func(uint64) *exec.Cmd
+}
+
+func NewSupervisor(executable string) (*Supervisor, error) {
+	if !filepath.IsAbs(executable) {
+		return nil, ErrUnavailable
+	}
+	return &Supervisor{command: func(epoch uint64) *exec.Cmd {
+		return exec.Command(executable, "--epoch", strconv.FormatUint(epoch, 10))
+	}}, nil
+}
+
+type keyCall struct {
+	name    string
+	pressed bool
+	result  chan error
+}
+
+// Client owns one fixed epoch. Cancel closes its independent lifetime pipe;
+// Done closes only after retirement was checked or its bounded wait expired.
+// Cancellation never kills the helper: it must remain alive to release keys.
+type Client struct {
+	cancel context.CancelFunc
+	ready  chan struct{}
+	done   chan struct{}
+	calls  chan keyCall
+	err    error // published by closing done
+}
+
+func (supervisor *Supervisor) Start(ctx context.Context, epoch uint64) (*Client, error) {
+	if ctx == nil || ctx.Err() != nil || epoch == 0 {
+		return nil, ErrUnavailable
+	}
+	supervisor.mu.Lock()
+	if supervisor.uncertain || supervisor.active {
+		supervisor.mu.Unlock()
+		return nil, ErrUnavailable
+	}
+	supervisor.active = true
+	supervisor.mu.Unlock()
+	owner, cancel := context.WithCancel(ctx)
+	client := &Client{cancel: cancel, ready: make(chan struct{}), done: make(chan struct{}), calls: make(chan keyCall)}
+	go func() {
+		err := supervisor.serve(owner, epoch, client)
+		cancel()
+		supervisor.mu.Lock()
+		supervisor.active = false
+		supervisor.uncertain = errors.Is(err, ErrUncertain)
+		supervisor.mu.Unlock()
+		client.err = err
+		close(client.done)
+	}()
+	select {
+	case <-client.ready:
+		if owner.Err() == nil {
+			return client, nil
+		}
+	case <-client.done:
+		return nil, client.err
+	case <-owner.Done():
+	}
+	cancel()
+	<-client.done
+	if client.err != nil {
+		return nil, client.err
+	}
+	return nil, ErrUnavailable
+}
+
+func (client *Client) Cancel()               { client.cancel() }
+func (client *Client) Done() <-chan struct{} { return client.done }
+
+// Wait reports cleanup certainty, not whether the last key reached an app.
+func (client *Client) Wait() error {
+	<-client.done
+	return client.err
+}
+
+// Apply has no retained queue. Cancellation after admission retires the entire
+// client rather than allowing an uncertain key transition to outlive its call.
+func (client *Client) Apply(ctx context.Context, name string, pressed bool) error {
+	if ctx == nil || ctx.Err() != nil || !validKey(name) {
+		return ErrUnavailable
+	}
+	call := keyCall{name: name, pressed: pressed, result: make(chan error, 1)}
+	select {
+	case <-client.done:
+		return ErrUnavailable
+	case <-ctx.Done():
+		return ErrUnavailable
+	case client.calls <- call:
+	}
+	select {
+	case err := <-call.result:
+		return err
+	case <-client.done:
+		return ErrUnavailable
+	case <-ctx.Done():
+		client.Cancel()
+		return ErrUnavailable
+	}
+}
+
+func validKey(name string) bool {
+	_, err := EncodeOperation(Operation{Kind: Down, Epoch: 1, Sequence: 1, Name: name})
+	return err == nil
+}
+
+type ackResult struct {
+	ack Ack
+	err error
+}
+
+func (supervisor *Supervisor) serve(ctx context.Context, epoch uint64, client *Client) error {
+	var files []*os.File
+	defer func() {
+		for _, file := range files {
+			file.Close()
+		}
+	}()
+	pipe := func() (*os.File, *os.File, error) {
+		read, write, err := os.Pipe()
+		if err == nil {
+			files = append(files, read, write)
+		}
+		return read, write, err
+	}
+	controlRead, control, err := pipe()
+	if err != nil {
+		return ErrUnavailable
+	}
+	output, outputWrite, err := pipe()
+	if err != nil {
+		return ErrUnavailable
+	}
+	lifetimeRead, lifetime, err := pipe()
+	if err != nil {
+		return ErrUnavailable
+	}
+	cmd := supervisor.command(epoch)
+	cmd.Stdin, cmd.Stdout, cmd.Stderr = controlRead, outputWrite, io.Discard
+	cmd.ExtraFiles = []*os.File{lifetimeRead}
+	// No CommandContext/Pdeathsig: writer death must allow native retirement.
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	if ctx.Err() != nil || cmd.Start() != nil {
+		return ErrUnavailable
+	}
+	controlRead.Close()
+	outputWrite.Close()
+	lifetimeRead.Close()
+	process := make(chan error, 1)
+	go func() { process <- cmd.Wait() }()
+	stopWorkers := make(chan struct{})
+	var workers sync.WaitGroup
+	workers.Add(2)
+	defer func() {
+		close(stopWorkers)
+		output.Close()
+		workers.Wait()
+	}()
+	go func() {
+		defer workers.Done()
+		select {
+		case <-ctx.Done():
+			lifetime.Close()
+		case <-stopWorkers:
+		}
+	}()
+	acks := make(chan ackResult, 1)
+	go func() {
+		defer workers.Done()
+		for {
+			var frame [AckSize]byte
+			_, err := io.ReadFull(output, frame[:])
+			var ack Ack
+			if err == nil {
+				ack, err = DecodeAck(frame[:])
+			}
+			select {
+			case acks <- ackResult{ack, err}:
+			case <-stopWorkers:
+				return
+			}
+			if err != nil {
+				return
+			}
+		}
+	}()
+	var sent, accepted uint64
+	var terminal *Ack
+	protocolFailed := false
+	streamEnded, readySeen := false, false
+	// Every return after launch goes through the independent cleanup barrier.
+	retire := func() error {
+		lifetime.Close()
+		control.Close()
+		timer := time.NewTimer(retirementTimeout)
+		defer timer.Stop()
+		var exited bool
+		var exitError error
+		for !exited || !streamEnded {
+			select {
+			case exitError = <-process:
+				exited = true
+			case result := <-acks:
+				if result.err != nil {
+					streamEnded = true
+					if result.err != io.EOF {
+						protocolFailed = true
+					}
+					continue
+				}
+				ack := result.ack
+				if terminal != nil || ack.Epoch != epoch {
+					protocolFailed = true
+				} else if !readySeen && ack == (Ack{Kind: Ready, Epoch: epoch}) {
+					readySeen = true
+				} else if ack.Kind == Accepted && ack.Sequence == sent && accepted < sent {
+					accepted = sent
+				} else if ack.Kind == Closed && ack.Sequence >= accepted && ack.Sequence <= sent {
+					terminal = &ack
+				} else {
+					protocolFailed = true
+				}
+			case <-timer.C:
+				// A blocked native server may leave a living helper. Never kill
+				// it or permit replacement; its eventual exit is still reaped.
+				return ErrUncertain
+			}
+		}
+		if protocolFailed || terminal == nil || terminal.CleanupFailed || exitError != nil {
+			return ErrUncertain
+		}
+		return nil
+	}
+	readAck := func() (Ack, error) {
+		timer := time.NewTimer(operationTimeout)
+		defer timer.Stop()
+		select {
+		case result := <-acks:
+			if result.err != nil {
+				streamEnded = true
+				if result.err != io.EOF {
+					protocolFailed = true
+				}
+			}
+			return result.ack, result.err
+		case <-ctx.Done():
+			return Ack{}, ErrUnavailable
+		case <-timer.C:
+			return Ack{}, ErrUnavailable
+		}
+	}
+	ready, err := readAck()
+	if err != nil || ready != (Ack{Kind: Ready, Epoch: epoch}) {
+		// An ACK consumed here cannot be reinterpreted as a valid terminal.
+		if err == nil {
+			protocolFailed = true
+		}
+		if cleanup := retire(); cleanup != nil {
+			return cleanup
+		}
+		return ErrUnavailable
+	}
+	readySeen = true
+	close(client.ready)
+	transact := func(kind byte, name string) error {
+		if ctx.Err() != nil || sent == math.MaxUint64 {
+			return ErrUnavailable
+		}
+		sent++
+		frame, err := EncodeOperation(Operation{Kind: kind, Epoch: epoch, Sequence: sent, Name: name})
+		if err != nil {
+			return ErrUnavailable
+		}
+		if err = control.SetWriteDeadline(time.Now().Add(operationTimeout)); err == nil {
+			err = writeAll(control, frame)
+		}
+		if err != nil {
+			return ErrUnavailable
+		}
+		ack, err := readAck()
+		if err != nil {
+			return ErrUnavailable
+		}
+		if ack == (Ack{Kind: Accepted, Epoch: epoch, Sequence: sent}) {
+			accepted = sent
+			return nil
+		}
+		if ack.Kind == Closed && ack.Epoch == epoch && ack.Sequence >= accepted && ack.Sequence <= sent {
+			terminal = &ack
+		} else {
+			protocolFailed = true
+		}
+		return ErrUnavailable
+	}
+	ticker := time.NewTicker(Watchdog / 3)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return retire()
+		case <-ticker.C:
+			if transact(Heartbeat, "") != nil {
+				return retire()
+			}
+		case call := <-client.calls:
+			kind := Up
+			if call.pressed {
+				kind = Down
+			}
+			err := transact(kind, call.name)
+			call.result <- err
+			if err != nil {
+				return retire()
+			}
+		case result := <-acks:
+			if result.err != nil {
+				streamEnded = true
+				protocolFailed = true
+			} else if result.ack.Kind == Closed && result.ack.Epoch == epoch && result.ack.Sequence == accepted {
+				terminal = &result.ack
+			} else {
+				protocolFailed = true
+			}
+			return retire()
+		}
+	}
+}

internal/remoteinput/supervisor_test.go

@@ -0,0 +1,284 @@
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package remoteinput
+
+import (
+	"context"
+	"errors"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"syscall"
+	"testing"
+	"time"
+)
+
+type supervisorTestTarget struct{ mode string }
+
+func (target supervisorTestTarget) Apply(string, bool) error {
+	switch target.mode {
+	case "exit":
+		os.Exit(0)
+	case "blocked":
+		// A timer models a native call that remains blocked without causing
+		// Go's non-race runtime to terminate the synthetic child as deadlocked.
+		time.Sleep(time.Hour)
+	}
+	return nil
+}
+
+func (target supervisorTestTarget) Close() error {
+	if target.mode == "cleanup-failure" {
+		return ErrCleanup
+	}
+	return nil
+}
+
+func TestSupervisorProcess(t *testing.T) {
+	mode := os.Getenv("CTHULHU_SUPERVISOR_TEST")
+	if mode == "" {
+		return
+	}
+	epoch, _ := strconv.ParseUint(os.Getenv("CTHULHU_SUPERVISOR_EPOCH"), 10, 64)
+	control, err := AdoptPipe(os.Stdin, false)
+	if err != nil {
+		os.Exit(2)
+	}
+	output, err := AdoptPipe(os.Stdout, true)
+	if err != nil {
+		os.Exit(2)
+	}
+	life, err := AdoptPipe(os.NewFile(3, "lifetime"), false)
+	if err != nil {
+		os.Exit(2)
+	}
+	if mode == "no-ready" || mode == "bad-ready" {
+		if mode == "bad-ready" {
+			frame, _ := EncodeAck(Ack{Kind: Ready, Epoch: epoch + 1})
+			writeAll(output, frame)
+		}
+		io.Copy(io.Discard, life)
+		frame, _ := EncodeAck(Ack{Kind: Closed, Reason: ReasonEOF, Epoch: epoch})
+		writeAll(output, frame)
+		os.Exit(0)
+	}
+	if mode == "wrong-epoch" || mode == "wrong-sequence" || mode == "duplicate-ready" || mode == "withheld" || mode == "bad-exit" {
+		ready, _ := EncodeAck(Ack{Kind: Ready, Epoch: epoch})
+		writeAll(output, ready)
+		var frame [OperationSize]byte
+		io.ReadFull(control, frame[:])
+		operation, _ := DecodeOperation(frame[:])
+		ack := Ack{Kind: Accepted, Epoch: epoch, Sequence: operation.Sequence}
+		switch mode {
+		case "wrong-epoch":
+			ack.Epoch++
+		case "wrong-sequence":
+			ack.Sequence++
+		case "duplicate-ready":
+			ack = Ack{Kind: Ready, Epoch: epoch}
+		}
+		if mode == "withheld" {
+			if os.WriteFile(os.Getenv("CTHULHU_SUPERVISOR_MARKER"), []byte("pending"), 0600) != nil {
+				os.Exit(2)
+			}
+		} else {
+			encoded, _ := EncodeAck(ack)
+			writeAll(output, encoded)
+		}
+		io.Copy(io.Discard, life)
+		closed, _ := EncodeAck(Ack{Kind: Closed, Reason: ReasonEOF, Epoch: epoch, Sequence: operation.Sequence})
+		writeAll(output, closed)
+		if mode == "bad-exit" {
+			os.Exit(1)
+		}
+		os.Exit(0)
+	}
+	err = Run(context.Background(), epoch, control, life, output, supervisorTestTarget{mode})
+	if err != nil {
+		os.Exit(1)
+	}
+	os.Exit(0)
+}
+
+func testSupervisor(t *testing.T, mode string) (*Supervisor, func() *exec.Cmd) {
+	t.Helper()
+	var command *exec.Cmd
+	supervisor := &Supervisor{command: func(epoch uint64) *exec.Cmd {
+		cmd := exec.Command(os.Args[0], "-test.run=^TestSupervisorProcess$")
+		cmd.Env = append(os.Environ(), "CTHULHU_SUPERVISOR_TEST="+mode,
+			"CTHULHU_SUPERVISOR_EPOCH="+strconv.FormatUint(epoch, 10))
+		t.Cleanup(func() {
+			if cmd.Process != nil {
+				cmd.Process.Kill()
+			}
+		})
+		command = cmd
+		return cmd
+	}}
+	return supervisor, func() *exec.Cmd { return command }
+}
+
+func startSupervisor(t *testing.T, supervisor *Supervisor, epoch uint64) *Client {
+	t.Helper()
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	client, err := supervisor.Start(ctx, epoch)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { client.Cancel(); client.Wait() })
+	return client
+}
+
+func TestSupervisorRetirementAndReplacement(t *testing.T) {
+	supervisor, command := testSupervisor(t, "normal")
+	client := startSupervisor(t, supervisor, 41)
+	group, err := syscall.Getpgid(command().Process.Pid)
+	if err != nil || group != command().Process.Pid || group == syscall.Getpgrp() {
+		t.Fatal("helper shares its writer's process group", err)
+	}
+	if _, err := supervisor.Start(context.Background(), 42); !errors.Is(err, ErrUnavailable) {
+		t.Fatal("overlapping helper admitted", err)
+	}
+	for _, pressed := range []bool{true, true, false} {
+		if err := client.Apply(t.Context(), "Shift_L", pressed); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// Heartbeats must retain an otherwise idle helper beyond its watchdog.
+	select {
+	case <-client.Done():
+		t.Fatal("idle helper retired", client.Wait())
+	case <-time.After(Watchdog + 100*time.Millisecond):
+	}
+	client.Cancel()
+	client.Cancel()
+	if err := client.Wait(); err != nil {
+		t.Fatal(err)
+	}
+	if err := client.Apply(t.Context(), "a", true); !errors.Is(err, ErrUnavailable) {
+		t.Fatal("retired helper accepted a key", err)
+	}
+	replacement := startSupervisor(t, supervisor, 42)
+	client.Cancel()
+	if err := replacement.Apply(t.Context(), "Control_R", true); err != nil {
+		t.Fatal("old cleanup affected replacement", err)
+	}
+	replacement.Cancel()
+	if err := replacement.Wait(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestSupervisorUncertainRetirementBlocksReplacement(t *testing.T) {
+	for _, mode := range []string{"cleanup-failure", "exit", "wrong-epoch", "wrong-sequence", "duplicate-ready", "blocked", "bad-exit"} {
+		t.Run(mode, func(t *testing.T) {
+			supervisor, command := testSupervisor(t, mode)
+			client := startSupervisor(t, supervisor, 1)
+			_ = client.Apply(t.Context(), "Shift_L", true)
+			client.Cancel()
+			if err := client.Wait(); !errors.Is(err, ErrUncertain) {
+				t.Fatal("missing cleanup barrier", err)
+			}
+			if _, err := supervisor.Start(t.Context(), 2); !errors.Is(err, ErrUnavailable) {
+				t.Fatal("uncertain input automatically replaced", err)
+			}
+			if mode == "blocked" && command().Process.Signal(syscall.Signal(0)) != nil {
+				t.Fatal("supervisor killed helper before native cleanup")
+			}
+		})
+	}
+}
+
+func TestSupervisorCancelledAdmittedCall(t *testing.T) {
+	marker := filepath.Join(t.TempDir(), "pending")
+	t.Setenv("CTHULHU_SUPERVISOR_MARKER", marker)
+	supervisor, _ := testSupervisor(t, "withheld")
+	client := startSupervisor(t, supervisor, 31)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	result := make(chan error, 1)
+	go func() { result <- client.Apply(ctx, "Shift_L", true) }()
+	deadline := time.NewTimer(operationTimeout)
+	defer deadline.Stop()
+	tick := time.NewTicker(5 * time.Millisecond)
+	defer tick.Stop()
+pending:
+	for {
+		select {
+		case <-tick.C:
+			if _, err := os.Stat(marker); err == nil {
+				break pending
+			}
+		case <-deadline.C:
+			t.Fatal("helper did not receive pending key")
+		}
+	}
+	if _, err := supervisor.Start(t.Context(), 32); !errors.Is(err, ErrUnavailable) {
+		t.Fatal("pending key owner replaced", err)
+	}
+	cancel()
+	if err := <-result; !errors.Is(err, ErrUnavailable) {
+		t.Fatal("cancelled uncertain key acknowledged", err)
+	}
+	if err := client.Wait(); err != nil {
+		t.Fatal("independent lifeline failed to retire pending call", err)
+	}
+}
+
+func TestSupervisorLaunchValidationAndFailure(t *testing.T) {
+	if _, err := NewSupervisor("relative-helper"); !errors.Is(err, ErrUnavailable) {
+		t.Fatal("relative executable accepted", err)
+	}
+	supervisor, _ := NewSupervisor(filepath.Join(t.TempDir(), "absent-helper"))
+	for _, epoch := range []uint64{0, 1, 2} {
+		if _, err := supervisor.Start(t.Context(), epoch); !errors.Is(err, ErrUnavailable) {
+			t.Fatal("launch failure not private", err)
+		}
+	}
+	if supervisor.active || supervisor.uncertain {
+		t.Fatal("failed exec poisoned unused target")
+	}
+}
+
+func TestSupervisorStartupBarrier(t *testing.T) {
+	for _, mode := range []string{"no-ready", "bad-ready"} {
+		t.Run(mode, func(t *testing.T) {
+			supervisor, _ := testSupervisor(t, mode)
+			_, err := supervisor.Start(t.Context(), 8)
+			if mode == "no-ready" {
+				if !errors.Is(err, ErrUnavailable) || supervisor.uncertain {
+					t.Fatal("confirmed startup retirement poisoned replacement", err)
+				}
+			} else if !errors.Is(err, ErrUncertain) || !supervisor.uncertain {
+				t.Fatal("invalid readiness bypassed retirement barrier", err)
+			}
+		})
+	}
+}
+
+func TestSupervisorContextAndRepresentation(t *testing.T) {
+	supervisor, _ := testSupervisor(t, "normal")
+	ctx, cancel := context.WithCancel(context.Background())
+	client, err := supervisor.Start(ctx, 11)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { cancel(); client.Wait() })
+	for _, name := range []string{"", "private unsupported name", "A"} {
+		if err := client.Apply(t.Context(), name, true); !errors.Is(err, ErrUnavailable) {
+			t.Fatal("invalid key admitted", err)
+		}
+	}
+	if err := client.Apply(t.Context(), "a", true); err != nil {
+		t.Fatal(err)
+	}
+	cancel()
+	if err := client.Wait(); err != nil {
+		t.Fatal("context cancellation prevented clean EOF retirement", err)
+	}
+}