fix(daemonmux): classify a missing remote daemon during the handshake

commit: bdf8e57f1106daaf6cf0a3507a70d975d66e3798
author: brice · 29d ago · 2026-09-23 08:22 UTC
parent: 26fe86a
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +67 -0

M internal/adapters/daemonmux/connector.go +18 -0


@@ -37,8 +37,10 @@ import (
37	37		"context"
38	38		"errors"
39	39		"fmt"
	40	+	"os/exec"
40	41		"sync"
41	42
	43	+	"github.com/bnema/vev/internal/adapters/sshstdio"
42	44		"github.com/bnema/vev/internal/domain"
43	45		"github.com/bnema/vev/internal/ports"
44	46		)

@@ -180,6 +182,9 @@ func dialError(err error) error {
180	182		if errors.Is(err, context.Canceled) \|\| errors.Is(err, context.DeadlineExceeded) {
181	183		return err
182	184		}
	185	+	if missingRemoteDaemon(err) {
	186	+	return ports.BrokerError{Code: ports.BrokerErrorNoDaemon, Cause: err}
	187	+	}
183	188		var typed ports.BrokerError
184	189		if errors.As(err, &typed) && typed.Validate() == nil {
185	190		return err

@@ -187,6 +192,16 @@ func dialError(err error) error {
187	192		return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Cause: err}
188	193		}
189	194
	195	+	// missingRemoteDaemon only accepts the reserved helper exit after SSH has
	196	+	// started; SSH's own failures (including exit 255) remain unavailable.
	197	+	func missingRemoteDaemon(err error) bool {
	198	+	if !errors.Is(err, sshstdio.ErrMuxSSHExit) {
	199	+	return false
	200	+	}
	201	+	var exit *exec.ExitError
	202	+	return errors.As(err, &exit) && exit.ExitCode() == sshstdio.MuxExitNoDaemon
	203	+	}
	204	+
190	205		// handshakeError classifies one handshake failure: cancellation and deadline
191	206		// stay recognizable, a peer refusal is a policy conflict, a local authority
192	207		// mismatch is an incompatibility, and anything else is unavailable.

@@ -197,6 +212,9 @@ func handshakeError(err error) error {
197	212		if errors.Is(err, context.Canceled) \|\| errors.Is(err, context.DeadlineExceeded) {
198	213		return err
199	214		}
	215	+	if missingRemoteDaemon(err) {
	216	+	return ports.BrokerError{Code: ports.BrokerErrorNoDaemon, Cause: err}
	217	+	}
200	218		var refusal *HandshakeRefusal
201	219		if errors.As(err, &refusal) {
202	220		return ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy, Cause: err}


@@ -37,8 +37,10 @@ import (
37	"context"	37		"context"
38	"errors"	38		"errors"
39	"fmt"	39		"fmt"
		40	+	"os/exec"
40	"sync"	41		"sync"
41		42
		43	+	"github.com/bnema/vev/internal/adapters/sshstdio"
42	"github.com/bnema/vev/internal/domain"	44		"github.com/bnema/vev/internal/domain"
43	"github.com/bnema/vev/internal/ports"	45		"github.com/bnema/vev/internal/ports"
44	)	46		)

@@ -180,6 +182,9 @@ func dialError(err error) error {
180	if errors.Is(err, context.Canceled) \|\| errors.Is(err, context.DeadlineExceeded) {	182		if errors.Is(err, context.Canceled) \|\| errors.Is(err, context.DeadlineExceeded) {
181	return err	183		return err
182	}	184		}
		185	+	if missingRemoteDaemon(err) {
		186	+	return ports.BrokerError{Code: ports.BrokerErrorNoDaemon, Cause: err}
		187	+	}
183	var typed ports.BrokerError	188		var typed ports.BrokerError
184	if errors.As(err, &typed) && typed.Validate() == nil {	189		if errors.As(err, &typed) && typed.Validate() == nil {
185	return err	190		return err

@@ -187,6 +192,16 @@ func dialError(err error) error {
187	return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Cause: err}	192		return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Cause: err}
188	}	193		}
189		194
		195	+	// missingRemoteDaemon only accepts the reserved helper exit after SSH has
		196	+	// started; SSH's own failures (including exit 255) remain unavailable.
		197	+	func missingRemoteDaemon(err error) bool {
		198	+	if !errors.Is(err, sshstdio.ErrMuxSSHExit) {
		199	+	return false
		200	+	}
		201	+	var exit *exec.ExitError
		202	+	return errors.As(err, &exit) && exit.ExitCode() == sshstdio.MuxExitNoDaemon
		203	+	}
		204	+
190	// handshakeError classifies one handshake failure: cancellation and deadline	205		// handshakeError classifies one handshake failure: cancellation and deadline
191	// stay recognizable, a peer refusal is a policy conflict, a local authority	206		// stay recognizable, a peer refusal is a policy conflict, a local authority
192	// mismatch is an incompatibility, and anything else is unavailable.	207		// mismatch is an incompatibility, and anything else is unavailable.

@@ -197,6 +212,9 @@ func handshakeError(err error) error {
197	if errors.Is(err, context.Canceled) \|\| errors.Is(err, context.DeadlineExceeded) {	212		if errors.Is(err, context.Canceled) \|\| errors.Is(err, context.DeadlineExceeded) {
198	return err	213		return err
199	}	214		}
		215	+	if missingRemoteDaemon(err) {
		216	+	return ports.BrokerError{Code: ports.BrokerErrorNoDaemon, Cause: err}
		217	+	}
200	var refusal *HandshakeRefusal	218		var refusal *HandshakeRefusal
201	if errors.As(err, &refusal) {	219		if errors.As(err, &refusal) {
202	return ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy, Cause: err}	220		return ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy, Cause: err}

M internal/adapters/daemonmux/connector_test.go +49 -0


@@ -2,7 +2,12 @@ package daemonmux
2	2
3	3		import (
4	4		"context"
	5	+	"errors"
	6	+	"fmt"
	7	+	"io"
5	8		"net"
	9	+	"os/exec"
	10	+	"runtime"
6	11		"sync"
7	12		"sync/atomic"
8	13		"testing"

@@ -12,6 +17,7 @@ import (
12	17
13	18		"github.com/bnema/vev/internal/adapters/clock"
14	19		"github.com/bnema/vev/internal/adapters/ipc"
	20	+	"github.com/bnema/vev/internal/adapters/sshstdio"
15	21		"github.com/bnema/vev/internal/domain"
16	22		"github.com/bnema/vev/internal/ports"
17	23		"github.com/bnema/vev/internal/protocol"

@@ -22,6 +28,49 @@ import (
22	28		// admission handshake, never a handshake budget.
23	29		const p3cTestDeadline = 5 * time.Second
24	30
	31	+	func TestDialErrorClassifiesMissingDaemonSeparately(t *testing.T) {
	32	+	if runtime.GOOS == "windows" {
	33	+	t.Skip("test uses the POSIX shell")
	34	+	}
	35	+	exitError := func(t *testing.T, code int) error {
	36	+	t.Helper()
	37	+	cmd := exec.Command("sh", "-c", fmt.Sprintf("exit %d", code))
	38	+	err := cmd.Run()
	39	+	var exit *exec.ExitError
	40	+	require.ErrorAs(t, err, &exit)
	41	+	return errors.Join(sshstdio.ErrMuxSSHExit, exit)
	42	+	}
	43	+
	44	+	tests := []struct {
	45	+	name string
	46	+	err error
	47	+	want ports.BrokerErrorCode
	48	+	}{
	49	+	{name: "reserved helper status means no daemon", err: exitError(t, sshstdio.MuxExitNoDaemon), want: ports.BrokerErrorNoDaemon},
	50	+	{name: "ssh transport status remains unavailable", err: exitError(t, 255), want: ports.BrokerErrorUnavailable},
	51	+	{name: "other helper status remains unavailable", err: exitError(t, 127), want: ports.BrokerErrorUnavailable},
	52	+	{name: "zero exit after carriage end remains unavailable", err: errors.Join(sshstdio.ErrMuxSSHExit, io.EOF), want: ports.BrokerErrorUnavailable},
	53	+	{name: "other mux failure remains unavailable", err: errors.New("socket timeout"), want: ports.BrokerErrorUnavailable},
	54	+	}
	55	+	for _, tt := range tests {
	56	+	t.Run(tt.name, func(t *testing.T) {
	57	+	for _, classify := range []struct {
	58	+	name string
	59	+	fn func(error) error
	60	+	}{
	61	+	{name: "dial", fn: dialError},
	62	+	{name: "handshake", fn: handshakeError},
	63	+	} {
	64	+	t.Run(classify.name, func(t *testing.T) {
	65	+	var brokerErr ports.BrokerError
	66	+	require.ErrorAs(t, classify.fn(tt.err), &brokerErr)
	67	+	require.Equal(t, tt.want, brokerErr.Code)
	68	+	})
	69	+	}
	70	+	})
	71	+	}
	72	+	}
	73	+
25	74		// rawCarriagePair returns two independent real raw framed transports joined by
26	75		// an in-memory duplex pipe. Each end frames through the shared streamframe
27	76		// carriage via the IPC adapter, so the bridge, physical preamble, pump, and


@@ -2,7 +2,12 @@ package daemonmux
2		2
3	import (	3		import (
4	"context"	4		"context"
		5	+	"errors"
		6	+	"fmt"
		7	+	"io"
5	"net"	8		"net"
		9	+	"os/exec"
		10	+	"runtime"
6	"sync"	11		"sync"
7	"sync/atomic"	12		"sync/atomic"
8	"testing"	13		"testing"

@@ -12,6 +17,7 @@ import (
12		17
13	"github.com/bnema/vev/internal/adapters/clock"	18		"github.com/bnema/vev/internal/adapters/clock"
14	"github.com/bnema/vev/internal/adapters/ipc"	19		"github.com/bnema/vev/internal/adapters/ipc"
		20	+	"github.com/bnema/vev/internal/adapters/sshstdio"
15	"github.com/bnema/vev/internal/domain"	21		"github.com/bnema/vev/internal/domain"
16	"github.com/bnema/vev/internal/ports"	22		"github.com/bnema/vev/internal/ports"
17	"github.com/bnema/vev/internal/protocol"	23		"github.com/bnema/vev/internal/protocol"

@@ -22,6 +28,49 @@ import (
22	// admission handshake, never a handshake budget.	28		// admission handshake, never a handshake budget.
23	const p3cTestDeadline = 5 * time.Second	29		const p3cTestDeadline = 5 * time.Second
24		30
		31	+	func TestDialErrorClassifiesMissingDaemonSeparately(t *testing.T) {
		32	+	if runtime.GOOS == "windows" {
		33	+	t.Skip("test uses the POSIX shell")
		34	+	}
		35	+	exitError := func(t *testing.T, code int) error {
		36	+	t.Helper()
		37	+	cmd := exec.Command("sh", "-c", fmt.Sprintf("exit %d", code))
		38	+	err := cmd.Run()
		39	+	var exit *exec.ExitError
		40	+	require.ErrorAs(t, err, &exit)
		41	+	return errors.Join(sshstdio.ErrMuxSSHExit, exit)
		42	+	}
		43	+
		44	+	tests := []struct {
		45	+	name string
		46	+	err error
		47	+	want ports.BrokerErrorCode
		48	+	}{
		49	+	{name: "reserved helper status means no daemon", err: exitError(t, sshstdio.MuxExitNoDaemon), want: ports.BrokerErrorNoDaemon},
		50	+	{name: "ssh transport status remains unavailable", err: exitError(t, 255), want: ports.BrokerErrorUnavailable},
		51	+	{name: "other helper status remains unavailable", err: exitError(t, 127), want: ports.BrokerErrorUnavailable},
		52	+	{name: "zero exit after carriage end remains unavailable", err: errors.Join(sshstdio.ErrMuxSSHExit, io.EOF), want: ports.BrokerErrorUnavailable},
		53	+	{name: "other mux failure remains unavailable", err: errors.New("socket timeout"), want: ports.BrokerErrorUnavailable},
		54	+	}
		55	+	for _, tt := range tests {
		56	+	t.Run(tt.name, func(t *testing.T) {
		57	+	for _, classify := range []struct {
		58	+	name string
		59	+	fn func(error) error
		60	+	}{
		61	+	{name: "dial", fn: dialError},
		62	+	{name: "handshake", fn: handshakeError},
		63	+	} {
		64	+	t.Run(classify.name, func(t *testing.T) {
		65	+	var brokerErr ports.BrokerError
		66	+	require.ErrorAs(t, classify.fn(tt.err), &brokerErr)
		67	+	require.Equal(t, tt.want, brokerErr.Code)
		68	+	})
		69	+	}
		70	+	})
		71	+	}
		72	+	}
		73	+
25	// rawCarriagePair returns two independent real raw framed transports joined by	74		// rawCarriagePair returns two independent real raw framed transports joined by
26	// an in-memory duplex pipe. Each end frames through the shared streamframe	75		// an in-memory duplex pipe. Each end frames through the shared streamframe
27	// carriage via the IPC adapter, so the bridge, physical preamble, pump, and	76		// carriage via the IPC adapter, so the bridge, physical preamble, pump, and