fix(agent): report the failures the channel paths dropped

CHANGELOG.md

@@ -8,6 +8,8 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Fixed
 
+- Three dropped failures. A reconnecting control host no longer leaks the connection it replaces — one descriptor per run, for the agent's whole life. An agent that cannot open its channel says so at the first ask instead of timing out on `no control host attached`. A local workdir that cannot be created fails the run, naming itself (#638).
+
 - `archive.extract-member` passes the member name to `tar` after `--`. A member named `-rf.txt` was read as options, so the observe saw a mismatch and the apply failed on every run. The name comes from whoever built the archive, not from the plan (#639).
 
 - A def written with the retired `pre-check` phase now gets the message telling it to rename, instead of the generic "expected a phase". The entry was keyed `check` — a valid phase, matched earlier — so it was unreachable, and its test asserted a substring the generic message already contained, so nothing said so (#637).

internal/agent/agent.go

@@ -33,10 +33,15 @@ func ServeOn(in io.Reader, out io.Writer, ex engine.Executor, sockDir string) er
 	var ch *Channel
 	if sockDir != "" {
 		c, err := Listen(sockDir)
-		if err == nil {
-			ch = c
-			defer func() { _ = ch.Close() }()
+		if err != nil {
+			// Not fatal, and not silent either: the run may declare a primitive it never
+			// reaches, so the job goes on — but with a channel that says why nobody will
+			// answer. Dropping the error left `no control host attached` after a 30s wait,
+			// naming the symptom and never the cause (#638).
+			c = Unavailable(err)
 		}
+		ch = c
+		defer func() { _ = ch.Close() }()
 	}
 	return write(out, runRequest(req, ex, ch))
 }

internal/agent/channel.go

@@ -49,6 +49,12 @@ type Channel struct {
 	// exercises against a container, which is the only place it can be proven.
 	child func(ex engine.Executor, args ...string) (string, error)
 
+	// openErr is set when the agent could not open its listener at all. The job still
+	// runs — a plan may declare a primitive it never reaches, and a `--dry-run` reaches
+	// none — but any ask fails naming this instead of timing out on a bridge that was
+	// never going to attach (#638).
+	openErr error
+
 	mu    sync.Mutex
 	conn  *proto.Conn
 	next  int
@@ -92,6 +98,13 @@ func (c *Channel) accept() {
 			continue
 		}
 		c.mu.Lock()
+		// The bridge being replaced is closed here. `drop()` is the only other closer and
+		// it runs when an *ask* discovers the connection is dead — so a control host that
+		// reconnects without the agent having asked anything in between left the old one
+		// open, one descriptor per reconnection, for as long as the agent lives (#638).
+		if c.conn != nil {
+			_ = c.conn.Close()
+		}
 		c.conn = conn
 		select {
 		case <-c.ready: // already armed by a previous bridge
@@ -104,7 +117,19 @@ func (c *Channel) accept() {
 	}
 }
 
-func (c *Channel) Close() error { return c.ln.Close() }
+func (c *Channel) Close() error {
+	if c.ln == nil { // an unavailable channel never listened
+		return nil
+	}
+	return c.ln.Close()
+}
+
+// Unavailable is the channel an agent gets when it could not listen. Every ask fails with
+// `err`, which beats both alternatives: a nil channel loses the cause, and failing the job
+// up front would break a run that declares a primitive and never reaches it.
+func Unavailable(err error) *Channel {
+	return &Channel{openErr: err, ready: make(chan struct{})}
+}
 
 // AskWith requests a resource from the control host and blocks until it answers.
 //
@@ -137,6 +162,9 @@ func (c *Channel) AskWith(resource string, payload []byte, vars map[string]strin
 // attached returns the live connection, waiting for a bridge if none has arrived yet.
 // Assumes c.mu is held, and releases it around the wait so accept() can install one.
 func (c *Channel) attached(resource string) (*proto.Conn, error) {
+	if c.openErr != nil {
+		return nil, fmt.Errorf("%s: this agent has no control channel: %w", resource, c.openErr)
+	}
 	if c.conn != nil {
 		return c.conn, nil
 	}

internal/agent/channel_leak_test.go

@@ -0,0 +1,82 @@
+package agent
+
+import (
+	"errors"
+	"net"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"shellf/internal/proto"
+)
+
+// attach dials the agent's socket and completes the handshake, handing back both ends of
+// the client side: the raw conn (for a read deadline) and the framed one. Unlike
+// `control`, it starts no reader goroutine — this test needs to observe what the agent
+// does to the connection, which a goroutine consuming it would hide.
+func attach(t *testing.T, sock string) (net.Conn, *proto.Conn) {
+	t.Helper()
+	raw, err := net.Dial("unix", sock)
+	if err != nil {
+		t.Fatal(err)
+	}
+	pc := proto.NewConn(raw)
+	if err := pc.Handshake(); err != nil {
+		t.Fatal(err)
+	}
+	return raw, pc
+}
+
+// waitConn blocks until the channel holds a connection other than `not`.
+func waitConn(t *testing.T, c *Channel, not *proto.Conn) *proto.Conn {
+	t.Helper()
+	for i := 0; i < 200; i++ {
+		c.mu.Lock()
+		cur := c.conn
+		c.mu.Unlock()
+		if cur != nil && cur != not {
+			return cur
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatal("no bridge attached in time")
+	return nil
+}
+
+// A reconnecting control host replaces the agent's connection. The replaced one must be
+// closed, or the agent leaks a descriptor per reconnection — and it lives up to two hours
+// (ADR-0005), so the ceiling is the number of runs in that window (#638).
+//
+// `drop()` closes, but only when an ask discovers the connection is dead. A run that asks
+// the control host for nothing between two bridges never takes that path, which is why
+// this went unseen.
+func TestChannel_ReplacedBridgeIsClosed(t *testing.T) {
+	wd := shortDir(t)
+	ch, err := Listen(wd)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = ch.Close() }()
+	sock := filepath.Join(wd, SockName)
+
+	raw1, pc1 := attach(t, sock)
+	defer func() { _ = raw1.Close() }()
+	first := waitConn(t, ch, nil)
+
+	// A second bridge attaches without the first having failed an ask.
+	raw2, _ := attach(t, sock)
+	defer func() { _ = raw2.Close() }()
+	waitConn(t, ch, first)
+
+	// The agent's end of the first bridge must be gone: reading from the client end
+	// returns EOF rather than blocking.
+	_ = raw1.SetReadDeadline(time.Now().Add(2 * time.Second))
+	if _, err := pc1.Recv(); err == nil {
+		t.Fatal("the replaced bridge answered: it was not closed")
+	} else {
+		var ne net.Error
+		if errors.As(err, &ne) && ne.Timeout() {
+			t.Fatal("the replaced connection is still open — accept() overwrote c.conn without closing it (#638)")
+		}
+	}
+}

internal/agent/channel_unavailable_test.go

@@ -0,0 +1,83 @@
+package agent
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"shellf/internal/proto"
+)
+
+// An agent that could not listen answers every ask with the reason (#638).
+//
+// Before, `ServeOn` and the resident loop both dropped the `Listen` error and ran with a
+// nil channel: a job that did ask waited out `attachWait` and failed with `no control host
+// attached` — the symptom, never the cause. The job still runs, because a plan may declare
+// a primitive it never reaches.
+func TestChannel_UnavailableCarriesTheCause(t *testing.T) {
+	cause := errors.New("bind: permission denied")
+	ch := Unavailable(cause)
+	defer func() { _ = ch.Close() }() // must not panic: it never listened
+
+	start := time.Now()
+	_, err := ch.AskWith("app.conf.j2", nil, nil)
+	if err == nil {
+		t.Fatal("an ask on an unavailable channel must fail")
+	}
+	if !strings.Contains(err.Error(), "app.conf.j2") {
+		t.Fatalf("the failure must name the resource: %v", err)
+	}
+	if !errors.Is(err, cause) {
+		t.Fatalf("the failure must carry why the channel could not open: %v", err)
+	}
+	// And it fails at once rather than sitting out the attach wait: nobody is coming.
+	if d := time.Since(start); d > time.Second {
+		t.Fatalf("an unavailable channel waited %v for a bridge that cannot attach", d)
+	}
+}
+
+// The wiring, not just the mechanism: `ServeOn` given a workdir it cannot listen in runs
+// the job anyway, and a step that asks the control host fails naming why (#638).
+//
+// The listen fails for a real reason here — the workdir is a *file*, so `net.Listen` on a
+// path inside it cannot bind — rather than by injecting an error, which would prove only
+// that the injection works.
+func TestServeOn_ListenFailureReachesTheStep(t *testing.T) {
+	notADir := filepath.Join(t.TempDir(), "workdir")
+	if err := os.WriteFile(notADir, []byte("not a directory"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	f := newComp()
+	body, _ := json.Marshal(proto.Request{
+		Mode: "apply",
+		Defs: map[string]string{
+			"deliver": `def deliver(src: str) { apply { out = ~file.read(src) shell { printf '%s' "$out" } return ok.done } }`,
+		},
+		Steps: []proto.Step{{Instruction: "deliver",
+			Args: map[string]string{"src": "/plan/conf.j2"}, Control: []string{"src"}}},
+	})
+	var out bytes.Buffer
+	if err := ServeOn(bytes.NewReader(body), &out, f, notADir); err != nil {
+		t.Fatal(err)
+	}
+	var resp proto.Response
+	if err := json.Unmarshal(out.Bytes(), &resp); err != nil {
+		t.Fatal(err)
+	}
+	if len(resp.Results) == 0 {
+		t.Fatal("the job must still run and report")
+	}
+	got := resp.Results[0].Shell
+	if resp.Results[0].Category != "err" || got == nil {
+		t.Fatalf("the step must fail: %+v", resp.Results[0])
+	}
+	if !strings.Contains(got.Stderr, "no control channel") {
+		t.Fatalf("the failure must say the agent has no channel, got: %q", got.Stderr)
+	}
+}

internal/agent/resident.go

@@ -37,10 +37,16 @@ func ServeResident(workdir, binPath string, ex engine.Executor, ttl time.Duratio
 	// created still runs every plan that asks nothing of the control host, which is
 	// almost all of them. Failing the agent outright would trade a working majority for
 	// a feature the plan may never use.
+	//
+	// Best-effort is not the same as silent, though, and this dropped `cherr` on the floor:
+	// a job that did ask waited out `attachWait` and failed with `no control host attached`,
+	// naming the symptom. `Unavailable` keeps the majority running and hands the minority
+	// the reason (#638).
 	ch, cherr := Listen(workdir)
-	if cherr == nil {
-		defer func() { _ = ch.Close() }()
+	if cherr != nil {
+		ch = Unavailable(cherr)
 	}
+	defer func() { _ = ch.Close() }()
 
 	last := time.Now()
 	for {

internal/transport/local.go

@@ -30,12 +30,17 @@ func (l Local) Run(agentBin string, req []byte) ([]byte, error) {
 	// something: a plan that does not keeps today's single-process behaviour.
 	if l.Channel != nil {
 		wd, err := os.MkdirTemp(sockBase(), "shellf-local")
-		if err == nil {
-			defer func() { _ = os.RemoveAll(wd) }()
-			args = append(args, wd)
-			stop := l.bridge(wd)
-			defer stop()
+		if err != nil {
+			// Reported, not skipped. The agent runs in another process and reads its
+			// workdir from argv, so there is no way to hand it the cause the way the
+			// agent's own `Unavailable` does — continuing here guarantees the operator
+			// sees `no control host channel available` and never why (#638).
+			return nil, fmt.Errorf("local agent: control channel workdir: %v", err)
 		}
+		defer func() { _ = os.RemoveAll(wd) }()
+		args = append(args, wd)
+		stop := l.bridge(wd)
+		defer stop()
 	}
 	cmd := exec.Command(agentBin, args...)
 	cmd.Stdin = bytes.NewReader(req)
@@ -49,7 +54,10 @@ func (l Local) Run(agentBin string, req []byte) ([]byte, error) {
 
 // sockBase prefers /dev/shm: a unix socket path is capped at ~108 bytes, and a long
 // TMPDIR would push past it with an error that reads like nonsense.
-func sockBase() string {
+//
+// A var so a test can point it at a path that cannot hold a directory: the failure above
+// is unreachable otherwise, and an unreachable branch is one nobody proves.
+var sockBase = func() string {
 	if fi, err := os.Stat("/dev/shm"); err == nil && fi.IsDir() {
 		return "/dev/shm"
 	}

internal/transport/local_workdir_test.go

@@ -0,0 +1,30 @@
+package transport
+
+import (
+	"io"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// A workdir that cannot be created is reported, not skipped (#638).
+//
+// It used to be `if err == nil { … }` with no else: the agent then started with no
+// workdir argument, opened no channel, and the first primitive failed with `no control
+// host channel available for this run` — a message about the target, for a failure on the
+// control host. The agent is another process and reads its workdir from argv, so unlike
+// the agent's own `Unavailable` there is no way to hand it the cause.
+func TestLocal_WorkdirThatCannotBeCreatedIsReported(t *testing.T) {
+	old := sockBase
+	sockBase = func() string { return filepath.Join(t.TempDir(), "no-such-parent") }
+	defer func() { sockBase = old }()
+
+	l := Local{Channel: func(io.Reader, io.WriteCloser) error { return nil }}
+	_, err := l.Run("/nonexistent/shellf", []byte(`{}`))
+	if err == nil {
+		t.Fatal("a workdir that cannot be created must fail the run")
+	}
+	if !strings.Contains(err.Error(), "control channel workdir") {
+		t.Fatalf("the failure must name what could not be made, got: %v", err)
+	}
+}