diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ec0563..3cf9098 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 2283fb3..492f0fa 100644 --- a/internal/agent/agent.go +++ b/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)) } diff --git a/internal/agent/channel.go b/internal/agent/channel.go index 5bd788b..70af6e2 100644 --- a/internal/agent/channel.go +++ b/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 } diff --git a/internal/agent/channel_leak_test.go b/internal/agent/channel_leak_test.go new file mode 100644 index 0000000..5f50f52 --- /dev/null +++ b/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)") + } + } +} diff --git a/internal/agent/channel_unavailable_test.go b/internal/agent/channel_unavailable_test.go new file mode 100644 index 0000000..4710de7 --- /dev/null +++ b/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) + } +} diff --git a/internal/agent/resident.go b/internal/agent/resident.go index 9b39b83..375bd84 100644 --- a/internal/agent/resident.go +++ b/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 { diff --git a/internal/transport/local.go b/internal/transport/local.go index b06a50f..b66eb39 100644 --- a/internal/transport/local.go +++ b/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" } diff --git a/internal/transport/local_workdir_test.go b/internal/transport/local_workdir_test.go new file mode 100644 index 0000000..42c5f09 --- /dev/null +++ b/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) + } +}