amadan

carlos / platform Public

activator: logListener.Close shuts the accepted connection and waits for the tee (plan 3, task 3 fix round 1)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 637a13ffa348bac0950c61d5858674d16e433aa5 parent 33d6062
2 files changed, +92 −7
  • internal/activator/loglistener.go +33 −7
  • internal/activator/loglistener_test.go +59 −0
diff --git a/internal/activator/loglistener.go b/internal/activator/loglistener.go
index 81c1eaf40..5d9bc6543 100644
--- a/internal/activator/loglistener.go
+++ b/internal/activator/loglistener.go
@@ -1,7 +1,6 @@
package activator
import (
- "bufio"
"io"
"net"
"os"
@@ -16,15 +15,21 @@ import (
// unit's StandardOutput=file: connects to this AF_UNIX socket once, at
// exec, and every line is teed to the agent's own stdout (journald) and to
// the app log sink, exactly as teeStd does for an exec child. One
-// connection per invocation is accepted; a second is closed unread. EOF
-// ends log collection and nothing else: it is never a sign the process is
-// gone (codex R1-4).
+// connection per listener is accepted; a second is closed unread. The
+// accepted flag never resets by design: a listener belongs to one
+// invocation, and a re-exec of the unit gets a fresh listener from
+// prepare. EOF ends log collection and nothing else: it is never a sign
+// the process is gone (codex R1-4).
type logListener struct {
ln net.Listener
emit func(string)
accepted atomic.Bool
closed atomic.Bool
- wg sync.WaitGroup
+ // conn is the one accepted connection, kept so Close can shut it and
+ // unblock the tee; wg covers the accept loop and the tee.
+ mu sync.Mutex
+ conn net.Conn
+ wg sync.WaitGroup
}
func listenLogs(path string, emit func(string)) (*logListener, error) {
@@ -49,29 +54,50 @@ func (l *logListener) accept() {
c.Close()
continue
}
+ l.mu.Lock()
+ if l.closed.Load() {
+ // Close ran between Accept and here; nothing may outlive it.
+ l.mu.Unlock()
+ c.Close()
+ continue
+ }
+ l.conn = c
+ l.wg.Add(1)
+ l.mu.Unlock()
go l.tee(c)
}
}
func (l *logListener) tee(c net.Conn) {
+ defer l.wg.Done()
defer c.Close()
var sink io.Writer = os.Stdout
if l.emit != nil {
sink = io.MultiWriter(os.Stdout, applog.NewLineWriter(execLogLineMax, l.emit))
}
- r := bufio.NewReaderSize(c, execLogLineMax)
- _, _ = io.Copy(sink, r)
+ // applog.LineWriter owns line assembly and the length cap, so no
+ // buffering layer is needed between the connection and the sink.
+ _, _ = io.Copy(sink, c)
}
// Exited always reports false: a log listener knows nothing about process
// state, and this method exists so a caller cannot mistake it for one.
func (l *logListener) Exited() bool { return false }
+// Close stops accepting, shuts the accepted connection so a blocked tee
+// returns, and waits for both goroutines. Nothing spawned by the listener
+// outlives Close: the process it belonged to is gone or being replaced,
+// and a late line must not reach the sink under the old invocation.
func (l *logListener) Close() error {
if l.closed.Swap(true) {
return nil
}
err := l.ln.Close()
+ l.mu.Lock()
+ if l.conn != nil {
+ l.conn.Close()
+ }
+ l.mu.Unlock()
l.wg.Wait()
return err
}
diff --git a/internal/activator/loglistener_test.go b/internal/activator/loglistener_test.go
index f1c25e73e..e9e1a7e4f 100644
--- a/internal/activator/loglistener_test.go
+++ b/internal/activator/loglistener_test.go
@@ -63,3 +63,62 @@ func TestLogListenerTeesOneConnectionAndRefusesASecond(t *testing.T) {
t.Error("EOF on the log connection must not mark the process exited")
}
}
+
+// Close must shut the accepted connection and wait for the tee: a line
+// written after Close returns must not reach the sink, and the tee
+// goroutine must not outlive the listener (task 3 review, item 1).
+func TestLogListenerCloseShutsTheConnectionAndWaitsForTheTee(t *testing.T) {
+ dir, err := os.MkdirTemp("", "ll")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ path := filepath.Join(dir, "h.run.sock")
+ var mu sync.Mutex
+ var lines []string
+ l, err := listenLogs(path, func(s string) { mu.Lock(); lines = append(lines, s); mu.Unlock() })
+ if err != nil {
+ t.Skipf("cannot bind here (%v); run with TMPDIR=/tmp/g outside the sandbox", err)
+ }
+ c, err := net.Dial("unix", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer c.Close()
+ if _, err := c.Write([]byte("before\n")); err != nil {
+ t.Fatal(err)
+ }
+ deadline := time.Now().Add(2 * time.Second)
+ for {
+ mu.Lock()
+ n := len(lines)
+ mu.Unlock()
+ if n == 1 || time.Now().After(deadline) {
+ break
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ done := make(chan error, 1)
+ go func() { done <- l.Close() }()
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatalf("Close: %v", err)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("Close blocked on the open connection instead of shutting it")
+ }
+ // The peer sees its end closed and a late line goes nowhere.
+ c.SetReadDeadline(time.Now().Add(time.Second))
+ if _, err := c.Read(make([]byte, 1)); err == nil {
+ t.Fatal("accepted connection still open after Close")
+ }
+ _, _ = c.Write([]byte("after\n"))
+ time.Sleep(50 * time.Millisecond)
+ mu.Lock()
+ got := strings.Join(lines, "|")
+ mu.Unlock()
+ if got != "before" {
+ t.Fatalf("lines after Close = %q, want only the line written before it", got)
+ }
+}