detached 1c8c669

fix(runtime): stabilize bounded shutdown and scanning

commit
1c8c669491aa09e09b9f7a167b4502d34a8d0952
author
brice · · 2026-07-26 09:08 UTC
parent
1f12e31
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
13 files · +420 -105
Unified
UnifiedSplit
M cmd/vev-perf-harness/hangup_test.go +123 -0
@@ -4,11 +4,13 @@ package main
44
55 import (
66 "encoding/json"
7+ "errors"
78 "fmt"
89 "os"
910 "os/exec"
1011 "path/filepath"
1112 "strings"
13+ "syscall"
1214 "testing"
1315 "time"
1416 )
@@ -133,6 +135,127 @@ func TestClientTracePairsAfterPTYHangup(t *testing.T) {
133135 }
134136 }
135137
138+// TestClientGracefulShutdownDrainsTracedStdioDescendant proves the launcher's
139+// normal client stop exits the attached shell before sending any signal. The
140+// client then closes and waits for its ssh transport, giving the traced _stdio
141+// descendant time to serialize its blocked receive's end mark and exit before
142+// the client is reaped.
143+func TestClientGracefulShutdownDrainsTracedStdioDescendant(t *testing.T) {
144+ if testing.Short() {
145+ t.Skip("public CLI stdio descendant lifecycle")
146+ }
147+ root, err := filepath.Abs(filepath.Join("..", ".."))
148+ if err != nil {
149+ t.Fatal(err)
150+ }
151+ runDir, err := os.MkdirTemp("", "vev-stdio-drain-")
152+ if err != nil {
153+ t.Fatal(err)
154+ }
155+ removeTestTree(t, runDir)
156+ bin := filepath.Join(runDir, "vev")
157+ build := exec.Command("go", "build", "-o", bin, "./")
158+ build.Dir = root
159+ if output, err := build.CombinedOutput(); err != nil {
160+ t.Fatalf("build public CLI: %v\n%s", err, output)
161+ }
162+
163+ s := scenario{ID: "stdio-drain", Roles: []string{"daemon", "ssh_stdio_peer", "client"}}
164+ maps := make([]processMapping, 0, len(s.Roles))
165+ for i, role := range s.Roles {
166+ id := fmt.Sprintf("%s-r001-%s-%02d", safeName(s.ID), safeName(role), i+1)
167+ path := filepath.Join(runDir, id+".jsonl")
168+ f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
169+ if err != nil {
170+ t.Fatal(err)
171+ }
172+ if err := f.Close(); err != nil {
173+ t.Fatal(err)
174+ }
175+ maps = append(maps, processMapping{ProcessID: id, ClockDomain: id, TracePath: path, Role: role, Scenario: s.ID, Run: 1, Identity: "harness-assigned:" + id})
176+ }
177+
178+ launcher := &cliLauncher{bin: bin}
179+ var daemon, peer launchedProcess
180+ var client *cliProcess
181+ var peerMapping processMapping
182+ for _, pm := range launchOrder(maps) {
183+ var command roleCommand
184+ switch pm.Role {
185+ case "daemon":
186+ command = roleCommand{Args: []string{"--daemon"}}
187+ case "ssh_stdio_peer":
188+ // An empty remote session requests an ephemeral session, avoiding a
189+ // separate bootstrap client while still executing the traced _stdio mode.
190+ command = roleCommand{Args: []string{"_stdio", ""}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}}
191+ case "client":
192+ command = roleCommand{Args: []string{"attach", "harness@127.0.0.1"}}
193+ }
194+ process, err := launcher.Launch(pm, command)
195+ if err != nil {
196+ t.Fatalf("launch %s: %v", pm.Role, err)
197+ }
198+ switch pm.Role {
199+ case "daemon":
200+ daemon = process
201+ if err := process.(interface{ WaitReady() error }).WaitReady(); err != nil {
202+ t.Fatalf("daemon readiness: %v", err)
203+ }
204+ case "ssh_stdio_peer":
205+ peer, peerMapping = process, pm
206+ case "client":
207+ client = process.(*cliProcess)
208+ }
209+ }
210+ if daemon == nil || peer == nil || client == nil {
211+ t.Fatal("did not launch complete stdio process tree")
212+ }
213+ defer func() {
214+ if err := peer.Close(); err != nil {
215+ t.Errorf("close peer role: %v", err)
216+ }
217+ if err := daemon.Close(); err != nil {
218+ t.Errorf("close daemon role: %v", err)
219+ }
220+ }()
221+ defer func() {
222+ if err := client.Close(); err != nil {
223+ t.Errorf("close client role: %v", err)
224+ }
225+ }()
226+
227+ if err := waitForInflightReceive(peerMapping.TracePath, 10*time.Second); err != nil {
228+ t.Fatalf("stdio descendant never reached blocked receive: %v", err)
229+ }
230+ processGroup := client.cmd.Process.Pid
231+ if err := client.Close(); err != nil {
232+ t.Fatalf("graceful client close: %v", err)
233+ }
234+ if _, err := mergeProcessTraces([]processMapping{peerMapping}); err != nil {
235+ t.Fatalf("stdio descendant trace did not drain: %v", err)
236+ }
237+ if err := waitForProcessGroupGone(processGroup, time.Second); err != nil {
238+ t.Fatal(err)
239+ }
240+}
241+
242+func waitForProcessGroupGone(pgid int, timeout time.Duration) error {
243+ deadline := time.Now().Add(timeout)
244+ for {
245+ err := syscall.Kill(-pgid, 0)
246+ if errors.Is(err, syscall.ESRCH) {
247+ return nil
248+ }
249+ if err != nil {
250+ return fmt.Errorf("probe process group %d: %w", pgid, err)
251+ }
252+ if time.Now().After(deadline) {
253+ return fmt.Errorf("process group %d remained after client shutdown", pgid)
254+ }
255+ time.Sleep(10 * time.Millisecond)
256+ }
257+}
258+
136259 // waitForInflightReceive polls a process trace until it holds an unpaired
137260 // adapter_receive_start (more starts than ends). It tolerates a partially
138261 // written trailing line while the traced process appends concurrently.
@@ -4,11 +4,13 @@ package main
4 4
5 import (5 import (
6 "encoding/json"6 "encoding/json"
7+ "errors"
7 "fmt"8 "fmt"
8 "os"9 "os"
9 "os/exec"10 "os/exec"
10 "path/filepath"11 "path/filepath"
11 "strings"12 "strings"
13+ "syscall"
12 "testing"14 "testing"
13 "time"15 "time"
14 )16 )
@@ -133,6 +135,127 @@ func TestClientTracePairsAfterPTYHangup(t *testing.T) {
133 }135 }
134 }136 }
135 137
138+// TestClientGracefulShutdownDrainsTracedStdioDescendant proves the launcher's
139+// normal client stop exits the attached shell before sending any signal. The
140+// client then closes and waits for its ssh transport, giving the traced _stdio
141+// descendant time to serialize its blocked receive's end mark and exit before
142+// the client is reaped.
143+func TestClientGracefulShutdownDrainsTracedStdioDescendant(t *testing.T) {
144+ if testing.Short() {
145+ t.Skip("public CLI stdio descendant lifecycle")
146+ }
147+ root, err := filepath.Abs(filepath.Join("..", ".."))
148+ if err != nil {
149+ t.Fatal(err)
150+ }
151+ runDir, err := os.MkdirTemp("", "vev-stdio-drain-")
152+ if err != nil {
153+ t.Fatal(err)
154+ }
155+ removeTestTree(t, runDir)
156+ bin := filepath.Join(runDir, "vev")
157+ build := exec.Command("go", "build", "-o", bin, "./")
158+ build.Dir = root
159+ if output, err := build.CombinedOutput(); err != nil {
160+ t.Fatalf("build public CLI: %v\n%s", err, output)
161+ }
162+
163+ s := scenario{ID: "stdio-drain", Roles: []string{"daemon", "ssh_stdio_peer", "client"}}
164+ maps := make([]processMapping, 0, len(s.Roles))
165+ for i, role := range s.Roles {
166+ id := fmt.Sprintf("%s-r001-%s-%02d", safeName(s.ID), safeName(role), i+1)
167+ path := filepath.Join(runDir, id+".jsonl")
168+ f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
169+ if err != nil {
170+ t.Fatal(err)
171+ }
172+ if err := f.Close(); err != nil {
173+ t.Fatal(err)
174+ }
175+ maps = append(maps, processMapping{ProcessID: id, ClockDomain: id, TracePath: path, Role: role, Scenario: s.ID, Run: 1, Identity: "harness-assigned:" + id})
176+ }
177+
178+ launcher := &cliLauncher{bin: bin}
179+ var daemon, peer launchedProcess
180+ var client *cliProcess
181+ var peerMapping processMapping
182+ for _, pm := range launchOrder(maps) {
183+ var command roleCommand
184+ switch pm.Role {
185+ case "daemon":
186+ command = roleCommand{Args: []string{"--daemon"}}
187+ case "ssh_stdio_peer":
188+ // An empty remote session requests an ephemeral session, avoiding a
189+ // separate bootstrap client while still executing the traced _stdio mode.
190+ command = roleCommand{Args: []string{"_stdio", ""}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}}
191+ case "client":
192+ command = roleCommand{Args: []string{"attach", "harness@127.0.0.1"}}
193+ }
194+ process, err := launcher.Launch(pm, command)
195+ if err != nil {
196+ t.Fatalf("launch %s: %v", pm.Role, err)
197+ }
198+ switch pm.Role {
199+ case "daemon":
200+ daemon = process
201+ if err := process.(interface{ WaitReady() error }).WaitReady(); err != nil {
202+ t.Fatalf("daemon readiness: %v", err)
203+ }
204+ case "ssh_stdio_peer":
205+ peer, peerMapping = process, pm
206+ case "client":
207+ client = process.(*cliProcess)
208+ }
209+ }
210+ if daemon == nil || peer == nil || client == nil {
211+ t.Fatal("did not launch complete stdio process tree")
212+ }
213+ defer func() {
214+ if err := peer.Close(); err != nil {
215+ t.Errorf("close peer role: %v", err)
216+ }
217+ if err := daemon.Close(); err != nil {
218+ t.Errorf("close daemon role: %v", err)
219+ }
220+ }()
221+ defer func() {
222+ if err := client.Close(); err != nil {
223+ t.Errorf("close client role: %v", err)
224+ }
225+ }()
226+
227+ if err := waitForInflightReceive(peerMapping.TracePath, 10*time.Second); err != nil {
228+ t.Fatalf("stdio descendant never reached blocked receive: %v", err)
229+ }
230+ processGroup := client.cmd.Process.Pid
231+ if err := client.Close(); err != nil {
232+ t.Fatalf("graceful client close: %v", err)
233+ }
234+ if _, err := mergeProcessTraces([]processMapping{peerMapping}); err != nil {
235+ t.Fatalf("stdio descendant trace did not drain: %v", err)
236+ }
237+ if err := waitForProcessGroupGone(processGroup, time.Second); err != nil {
238+ t.Fatal(err)
239+ }
240+}
241+
242+func waitForProcessGroupGone(pgid int, timeout time.Duration) error {
243+ deadline := time.Now().Add(timeout)
244+ for {
245+ err := syscall.Kill(-pgid, 0)
246+ if errors.Is(err, syscall.ESRCH) {
247+ return nil
248+ }
249+ if err != nil {
250+ return fmt.Errorf("probe process group %d: %w", pgid, err)
251+ }
252+ if time.Now().After(deadline) {
253+ return fmt.Errorf("process group %d remained after client shutdown", pgid)
254+ }
255+ time.Sleep(10 * time.Millisecond)
256+ }
257+}
258+
136 // waitForInflightReceive polls a process trace until it holds an unpaired259 // waitForInflightReceive polls a process trace until it holds an unpaired
137 // adapter_receive_start (more starts than ends). It tolerates a partially260 // adapter_receive_start (more starts than ends). It tolerates a partially
138 // written trailing line while the traced process appends concurrently.261 // written trailing line while the traced process appends concurrently.
M cmd/vev-perf-harness/launcher.go +35 -37
@@ -117,21 +117,22 @@ type terminalOutput interface {
117117 }
118118
119119 type cliProcess struct {
120- cmd *exec.Cmd
121- pty io.ReadWriteCloser
122- output terminalOutput
123- chunks chan []byte
124- done chan struct{}
125- closed sync.Once
126- waitErr chan error
127- waitTimeout func() <-chan time.Time
128- forceCleanup func()
129- forceKill func()
130- shutdown func() error
131- cleanupRuntime func() error
132- terminalReady bool
133- readyPath string
134- closeResult error
120+ cmd *exec.Cmd
121+ pty io.ReadWriteCloser
122+ output terminalOutput
123+ chunks chan []byte
124+ done chan struct{}
125+ closed sync.Once
126+ waitErr chan error
127+ waitTimeout func() <-chan time.Time
128+ gracefulShutdown func()
129+ forceCleanup func()
130+ forceKill func()
131+ shutdown func() error
132+ cleanupRuntime func() error
133+ terminalReady bool
134+ readyPath string
135+ closeResult error
135136 }
136137
137138 func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) {
@@ -203,6 +204,12 @@ func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProces
203204 }
204205 _ = slave.Close()
205206 p.pty, p.output, p.chunks, p.done = master, output, make(chan []byte, 32), make(chan struct{})
207+ // Stop workload generation through the attached shell before any signal.
208+ // A normal shell exit makes the daemon detach the client; the client then
209+ // closes and waits for its ssh transport, allowing a traced _stdio
210+ // descendant to flush end marks and exit. Process-group signals are reserved
211+ // for timeout escalation because they would kill the descendant concurrently.
212+ p.gracefulShutdown = func() { _, _ = master.Write([]byte("exit\n")) }
206213 go p.copyTerminal()
207214 } else {
208215 // Every launched role owns a dedicated process group, so the bounded
@@ -629,11 +636,10 @@ func (e *ptyLocalEcho) applicationOutput(chunk []byte) []byte {
629636 // Close tears down a launched role along a bounded escalation ladder. Each rung
630637 // performs one action, then waits a single closeDeadline for the process to be
631638 // reaped before the next rung escalates. The graceful rungs — the public daemon
632-// stop, the client shell exit, the controlling-terminal hangup, and SIGTERM —
633-// let a role close its blocked transport operations and serialize their end
634-// marks; only a SIGKILL severs a process before it can, so only that rung fails
635-// teardown. Every path closes the PTY exactly once and finalizes output and
636-// runtime state.
639+// stop and a client shell exit — let a role close its blocked transport
640+// operations and drain descendants' end marks. Process-group SIGTERM is reserved
641+// for timeout escalation; only a SIGKILL rung fails teardown. Every path closes
642+// the PTY exactly once and finalizes output and runtime state.
637643 func (p *cliProcess) Close() error {
638644 p.closed.Do(func() {
639645 // 1. Drain terminal output for the whole teardown. Without an active
@@ -682,27 +688,19 @@ func (p *cliProcess) Close() error {
682688 _ = p.shutdown()
683689 waitForExit()
684690 }
685- // 3. Client role: a public terminal client owns a shell session. Ask that
686- // shell to exit so the client detaches and its transport finishes its
687- // receive span cleanly.
688- if p.pty != nil && !reaped() {
689- _, _ = p.pty.Write([]byte("exit\n"))
690- waitForExit()
691- }
692- // 4. Hang up the controlling terminal. The attach path installs a signal
693- // handler, so this SIGHUP is a graceful detach — not an abrupt kill — and
694- // the client still flushes its in-flight receive end mark.
695- if p.pty != nil && !reaped() {
696- closePTY()
691+ // 3. Client role: stop the shell normally. Session teardown detaches the
692+ // client, which closes and waits for ssh/_stdio before it exits.
693+ if p.gracefulShutdown != nil && !reaped() {
694+ p.gracefulShutdown()
697695 waitForExit()
698696 }
699- // 5. SIGTERM the role's own process group. The daemon and, through its
700- // attach signal handler, the client both treat this as a graceful stop.
697+ // 4. Escalate a role that ignored graceful shutdown to process-group
698+ // SIGTERM. This remains bounded and cleans up hung descendants.
701699 if !reaped() {
702700 p.forceProcessGroupCleanup()
703701 waitForExit()
704702 }
705- // 6. SIGKILL the role's process group. This severs the process; an
703+ // 5. SIGKILL the role's process group. This severs the process; an
706704 // in-flight observed operation cannot be closed, so its trace may truncate.
707705 killed := false
708706 if !reaped() {
@@ -710,12 +708,12 @@ func (p *cliProcess) Close() error {
710708 p.forceProcessGroupKill()
711709 waitForExit()
712710 }
713- // 7. A SIGKILL escalation cannot guarantee a complete trace. Fail teardown
711+ // 6. A SIGKILL escalation cannot guarantee a complete trace. Fail teardown
714712 // with a clear cause rather than letting the truncation surface later as a
715713 // downstream merge mystery. The graceful rungs never error on their own.
716714 if killed {
717715 if exited {
718- p.closeResult = errors.New("public CLI process did not exit after hangup and SIGTERM; killed")
716+ p.closeResult = errors.New("public CLI process did not exit after graceful shutdown and SIGTERM; killed")
719717 } else {
720718 p.closeResult = errors.New("timed out reaping public CLI process")
721719 }
@@ -117,21 +117,22 @@ type terminalOutput interface {
117 }117 }
118 118
119 type cliProcess struct {119 type cliProcess struct {
120- cmd *exec.Cmd120+ cmd *exec.Cmd
121- pty io.ReadWriteCloser121+ pty io.ReadWriteCloser
122- output terminalOutput122+ output terminalOutput
123- chunks chan []byte123+ chunks chan []byte
124- done chan struct{}124+ done chan struct{}
125- closed sync.Once125+ closed sync.Once
126- waitErr chan error126+ waitErr chan error
127- waitTimeout func() <-chan time.Time127+ waitTimeout func() <-chan time.Time
128- forceCleanup func()128+ gracefulShutdown func()
129- forceKill func()129+ forceCleanup func()
130- shutdown func() error130+ forceKill func()
131- cleanupRuntime func() error131+ shutdown func() error
132- terminalReady bool132+ cleanupRuntime func() error
133- readyPath string133+ terminalReady bool
134- closeResult error134+ readyPath string
135+ closeResult error
135 }136 }
136 137
137 func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) {138 func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) {
@@ -203,6 +204,12 @@ func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProces
203 }204 }
204 _ = slave.Close()205 _ = slave.Close()
205 p.pty, p.output, p.chunks, p.done = master, output, make(chan []byte, 32), make(chan struct{})206 p.pty, p.output, p.chunks, p.done = master, output, make(chan []byte, 32), make(chan struct{})
207+ // Stop workload generation through the attached shell before any signal.
208+ // A normal shell exit makes the daemon detach the client; the client then
209+ // closes and waits for its ssh transport, allowing a traced _stdio
210+ // descendant to flush end marks and exit. Process-group signals are reserved
211+ // for timeout escalation because they would kill the descendant concurrently.
212+ p.gracefulShutdown = func() { _, _ = master.Write([]byte("exit\n")) }
206 go p.copyTerminal()213 go p.copyTerminal()
207 } else {214 } else {
208 // Every launched role owns a dedicated process group, so the bounded215 // Every launched role owns a dedicated process group, so the bounded
@@ -629,11 +636,10 @@ func (e *ptyLocalEcho) applicationOutput(chunk []byte) []byte {
629 // Close tears down a launched role along a bounded escalation ladder. Each rung636 // Close tears down a launched role along a bounded escalation ladder. Each rung
630 // performs one action, then waits a single closeDeadline for the process to be637 // performs one action, then waits a single closeDeadline for the process to be
631 // reaped before the next rung escalates. The graceful rungs — the public daemon638 // reaped before the next rung escalates. The graceful rungs — the public daemon
632-// stop, the client shell exit, the controlling-terminal hangup, and SIGTERM —639+// stop and a client shell exit — let a role close its blocked transport
633-// let a role close its blocked transport operations and serialize their end640+// operations and drain descendants' end marks. Process-group SIGTERM is reserved
634-// marks; only a SIGKILL severs a process before it can, so only that rung fails641+// for timeout escalation; only a SIGKILL rung fails teardown. Every path closes
635-// teardown. Every path closes the PTY exactly once and finalizes output and642+// the PTY exactly once and finalizes output and runtime state.
636-// runtime state.
637 func (p *cliProcess) Close() error {643 func (p *cliProcess) Close() error {
638 p.closed.Do(func() {644 p.closed.Do(func() {
639 // 1. Drain terminal output for the whole teardown. Without an active645 // 1. Drain terminal output for the whole teardown. Without an active
@@ -682,27 +688,19 @@ func (p *cliProcess) Close() error {
682 _ = p.shutdown()688 _ = p.shutdown()
683 waitForExit()689 waitForExit()
684 }690 }
685- // 3. Client role: a public terminal client owns a shell session. Ask that691+ // 3. Client role: stop the shell normally. Session teardown detaches the
686- // shell to exit so the client detaches and its transport finishes its692+ // client, which closes and waits for ssh/_stdio before it exits.
687- // receive span cleanly.693+ if p.gracefulShutdown != nil && !reaped() {
688- if p.pty != nil && !reaped() {694+ p.gracefulShutdown()
689- _, _ = p.pty.Write([]byte("exit\n"))
690- waitForExit()
691- }
692- // 4. Hang up the controlling terminal. The attach path installs a signal
693- // handler, so this SIGHUP is a graceful detach — not an abrupt kill — and
694- // the client still flushes its in-flight receive end mark.
695- if p.pty != nil && !reaped() {
696- closePTY()
697 waitForExit()695 waitForExit()
698 }696 }
699- // 5. SIGTERM the role's own process group. The daemon and, through its697+ // 4. Escalate a role that ignored graceful shutdown to process-group
700- // attach signal handler, the client both treat this as a graceful stop.698+ // SIGTERM. This remains bounded and cleans up hung descendants.
701 if !reaped() {699 if !reaped() {
702 p.forceProcessGroupCleanup()700 p.forceProcessGroupCleanup()
703 waitForExit()701 waitForExit()
704 }702 }
705- // 6. SIGKILL the role's process group. This severs the process; an703+ // 5. SIGKILL the role's process group. This severs the process; an
706 // in-flight observed operation cannot be closed, so its trace may truncate.704 // in-flight observed operation cannot be closed, so its trace may truncate.
707 killed := false705 killed := false
708 if !reaped() {706 if !reaped() {
@@ -710,12 +708,12 @@ func (p *cliProcess) Close() error {
710 p.forceProcessGroupKill()708 p.forceProcessGroupKill()
711 waitForExit()709 waitForExit()
712 }710 }
713- // 7. A SIGKILL escalation cannot guarantee a complete trace. Fail teardown711+ // 6. A SIGKILL escalation cannot guarantee a complete trace. Fail teardown
714 // with a clear cause rather than letting the truncation surface later as a712 // with a clear cause rather than letting the truncation surface later as a
715 // downstream merge mystery. The graceful rungs never error on their own.713 // downstream merge mystery. The graceful rungs never error on their own.
716 if killed {714 if killed {
717 if exited {715 if exited {
718- p.closeResult = errors.New("public CLI process did not exit after hangup and SIGTERM; killed")716+ p.closeResult = errors.New("public CLI process did not exit after graceful shutdown and SIGTERM; killed")
719 } else {717 } else {
720 p.closeResult = errors.New("timed out reaping public CLI process")718 p.closeResult = errors.New("timed out reaping public CLI process")
721 }719 }
M cmd/vev-perf-harness/launcher_test.go +45 -34
@@ -27,29 +27,22 @@ func (discardTerminalOutput) Close() error { return nil }
2727 func TestCLIProcessCloseDrainsQueuedTerminalOutputBeforeClientExit(t *testing.T) {
2828 harnessPTY, clientPTY := net.Pipe()
2929 t.Cleanup(func() { _ = clientPTY.Close() })
30- waitErr := make(chan error)
30+ waitErr := make(chan error, 1)
3131 secondRead := make(chan struct{})
3232 p := &cliProcess{
33- pty: harnessPTY,
34- output: discardTerminalOutput{},
35- chunks: make(chan []byte, 1),
36- done: make(chan struct{}),
37- waitErr: waitErr,
38- waitTimeout: func() <-chan time.Time { return time.After(time.Second) },
33+ pty: harnessPTY,
34+ output: discardTerminalOutput{},
35+ chunks: make(chan []byte, 1),
36+ done: make(chan struct{}),
37+ waitErr: waitErr,
38+ waitTimeout: func() <-chan time.Time { return time.After(time.Second) },
39+ forceCleanup: func() { waitErr <- nil },
3940 }
4041 go p.copyTerminal()
4142 go func() {
4243 _, _ = clientPTY.Write([]byte("first"))
4344 _, _ = clientPTY.Write([]byte("second"))
4445 close(secondRead)
45- buf := make([]byte, len("exit\n"))
46- if _, err := clientPTY.Read(buf); err != nil {
47- return
48- }
49- if _, err := clientPTY.Write([]byte("third")); err != nil {
50- return
51- }
52- waitErr <- nil
5346 }()
5447 select {
5548 case <-secondRead:
@@ -134,6 +127,26 @@ func TestCLIProcessCloseIsBoundedWhenPublicRoleDoesNotExit(t *testing.T) {
134127 }
135128 }
136129
130+func TestCLIProcessCloseEscalatesHungClientAfterGracefulDrainTimeout(t *testing.T) {
131+ deadline := make(chan time.Time)
132+ close(deadline)
133+ var graceful, term, kill int
134+ p := &cliProcess{
135+ pty: &orderedPTY{close: func() {}},
136+ waitErr: make(chan error),
137+ waitTimeout: func() <-chan time.Time { return deadline },
138+ gracefulShutdown: func() { graceful++ },
139+ forceCleanup: func() { term++ },
140+ forceKill: func() { kill++ },
141+ }
142+ if err := p.Close(); err == nil {
143+ t.Fatal("hung client close succeeded")
144+ }
145+ if graceful != 1 || term != 1 || kill != 1 {
146+ t.Fatalf("graceful/TERM/KILL calls = %d/%d/%d, want 1/1/1", graceful, term, kill)
147+ }
148+}
149+
137150 func TestCLIProcessCloseEscalationOutcome(t *testing.T) {
138151 tests := []struct {
139152 name string
@@ -252,6 +265,11 @@ func TestCLILauncherIsolatesRoleXDGEnvironmentAcrossRepetitions(t *testing.T) {
252265 if err != nil {
253266 t.Fatalf("run %d %s launch: %v", run, role, err)
254267 }
268+ // The fixture exits on its own. Synchronize with that exit instead of
269+ // letting Close race its SIGTERM against the environment capture.
270+ proc := p.(*cliProcess)
271+ exitErr := <-proc.waitErr
272+ proc.waitErr <- exitErr
255273 runtimeDir := launcher.runtimes[runDir]
256274 if err := p.Close(); err != nil {
257275 t.Fatalf("run %d %s close: %v", run, role, err)
@@ -383,36 +401,29 @@ func TestTraceEnvironmentMatchesProcessMapping(t *testing.T) {
383401 }
384402 }
385403
386-func TestCLIProcessCloseGracefulDetachCompletesSpanBeforeForcedCleanup(t *testing.T) {
404+func TestCLIProcessCloseStopsShellBeforeProcessGroupCleanup(t *testing.T) {
387405 var order []string
388406 waitErr := make(chan error, 1)
389- timeout := make(chan time.Time)
390407 p := &cliProcess{
391- pty: &orderedPTY{
392- write: func(input []byte) {
393- if string(input) != "exit\n" {
394- t.Fatalf("graceful exit input=%q", input)
395- }
396- order = append(order, "exit_written")
397- // The graceful client exit has closed its transport and emitted
398- // its adapter receive-end mark before cmd.Wait completes.
399- order = append(order, "adapter_receive_end")
400- waitErr <- nil
401- },
402- close: func() { order = append(order, "pty_closed") },
408+ pty: &orderedPTY{close: func() { order = append(order, "pty_closed") }},
409+ output: &orderedOutput{close: func() { order = append(order, "output_closed") }},
410+ waitErr: waitErr,
411+ gracefulShutdown: func() {
412+ order = append(order, "shell_exit")
413+ // The client closes and waits for its transport descendant before it
414+ // reports its own exit to the launcher.
415+ order = append(order, "descendant_receive_end")
416+ waitErr <- nil
403417 },
404- output: &orderedOutput{close: func() { order = append(order, "output_closed") }},
405- waitErr: waitErr,
406- waitTimeout: func() <-chan time.Time { return timeout },
407418 forceCleanup: func() {
408- order = append(order, "forced_process_group_cleanup")
419+ t.Fatal("normal shutdown sent SIGTERM to the traced descendant process group")
409420 },
410421 }
411422
412423 if err := p.Close(); err != nil {
413424 t.Fatal(err)
414425 }
415- if !equalStrings(order, []string{"exit_written", "adapter_receive_end", "pty_closed", "output_closed"}) {
426+ if !equalStrings(order, []string{"shell_exit", "descendant_receive_end", "pty_closed", "output_closed"}) {
416427 t.Fatalf("cleanup order=%q", order)
417428 }
418429 }
@@ -27,29 +27,22 @@ func (discardTerminalOutput) Close() error { return nil }
27 func TestCLIProcessCloseDrainsQueuedTerminalOutputBeforeClientExit(t *testing.T) {27 func TestCLIProcessCloseDrainsQueuedTerminalOutputBeforeClientExit(t *testing.T) {
28 harnessPTY, clientPTY := net.Pipe()28 harnessPTY, clientPTY := net.Pipe()
29 t.Cleanup(func() { _ = clientPTY.Close() })29 t.Cleanup(func() { _ = clientPTY.Close() })
30- waitErr := make(chan error)30+ waitErr := make(chan error, 1)
31 secondRead := make(chan struct{})31 secondRead := make(chan struct{})
32 p := &cliProcess{32 p := &cliProcess{
33- pty: harnessPTY,33+ pty: harnessPTY,
34- output: discardTerminalOutput{},34+ output: discardTerminalOutput{},
35- chunks: make(chan []byte, 1),35+ chunks: make(chan []byte, 1),
36- done: make(chan struct{}),36+ done: make(chan struct{}),
37- waitErr: waitErr,37+ waitErr: waitErr,
38- waitTimeout: func() <-chan time.Time { return time.After(time.Second) },38+ waitTimeout: func() <-chan time.Time { return time.After(time.Second) },
39+ forceCleanup: func() { waitErr <- nil },
39 }40 }
40 go p.copyTerminal()41 go p.copyTerminal()
41 go func() {42 go func() {
42 _, _ = clientPTY.Write([]byte("first"))43 _, _ = clientPTY.Write([]byte("first"))
43 _, _ = clientPTY.Write([]byte("second"))44 _, _ = clientPTY.Write([]byte("second"))
44 close(secondRead)45 close(secondRead)
45- buf := make([]byte, len("exit\n"))
46- if _, err := clientPTY.Read(buf); err != nil {
47- return
48- }
49- if _, err := clientPTY.Write([]byte("third")); err != nil {
50- return
51- }
52- waitErr <- nil
53 }()46 }()
54 select {47 select {
55 case <-secondRead:48 case <-secondRead:
@@ -134,6 +127,26 @@ func TestCLIProcessCloseIsBoundedWhenPublicRoleDoesNotExit(t *testing.T) {
134 }127 }
135 }128 }
136 129
130+func TestCLIProcessCloseEscalatesHungClientAfterGracefulDrainTimeout(t *testing.T) {
131+ deadline := make(chan time.Time)
132+ close(deadline)
133+ var graceful, term, kill int
134+ p := &cliProcess{
135+ pty: &orderedPTY{close: func() {}},
136+ waitErr: make(chan error),
137+ waitTimeout: func() <-chan time.Time { return deadline },
138+ gracefulShutdown: func() { graceful++ },
139+ forceCleanup: func() { term++ },
140+ forceKill: func() { kill++ },
141+ }
142+ if err := p.Close(); err == nil {
143+ t.Fatal("hung client close succeeded")
144+ }
145+ if graceful != 1 || term != 1 || kill != 1 {
146+ t.Fatalf("graceful/TERM/KILL calls = %d/%d/%d, want 1/1/1", graceful, term, kill)
147+ }
148+}
149+
137 func TestCLIProcessCloseEscalationOutcome(t *testing.T) {150 func TestCLIProcessCloseEscalationOutcome(t *testing.T) {
138 tests := []struct {151 tests := []struct {
139 name string152 name string
@@ -252,6 +265,11 @@ func TestCLILauncherIsolatesRoleXDGEnvironmentAcrossRepetitions(t *testing.T) {
252 if err != nil {265 if err != nil {
253 t.Fatalf("run %d %s launch: %v", run, role, err)266 t.Fatalf("run %d %s launch: %v", run, role, err)
254 }267 }
268+ // The fixture exits on its own. Synchronize with that exit instead of
269+ // letting Close race its SIGTERM against the environment capture.
270+ proc := p.(*cliProcess)
271+ exitErr := <-proc.waitErr
272+ proc.waitErr <- exitErr
255 runtimeDir := launcher.runtimes[runDir]273 runtimeDir := launcher.runtimes[runDir]
256 if err := p.Close(); err != nil {274 if err := p.Close(); err != nil {
257 t.Fatalf("run %d %s close: %v", run, role, err)275 t.Fatalf("run %d %s close: %v", run, role, err)
@@ -383,36 +401,29 @@ func TestTraceEnvironmentMatchesProcessMapping(t *testing.T) {
383 }401 }
384 }402 }
385 403
386-func TestCLIProcessCloseGracefulDetachCompletesSpanBeforeForcedCleanup(t *testing.T) {404+func TestCLIProcessCloseStopsShellBeforeProcessGroupCleanup(t *testing.T) {
387 var order []string405 var order []string
388 waitErr := make(chan error, 1)406 waitErr := make(chan error, 1)
389- timeout := make(chan time.Time)
390 p := &cliProcess{407 p := &cliProcess{
391- pty: &orderedPTY{408+ pty: &orderedPTY{close: func() { order = append(order, "pty_closed") }},
392- write: func(input []byte) {409+ output: &orderedOutput{close: func() { order = append(order, "output_closed") }},
393- if string(input) != "exit\n" {410+ waitErr: waitErr,
394- t.Fatalf("graceful exit input=%q", input)411+ gracefulShutdown: func() {
395- }412+ order = append(order, "shell_exit")
396- order = append(order, "exit_written")413+ // The client closes and waits for its transport descendant before it
397- // The graceful client exit has closed its transport and emitted414+ // reports its own exit to the launcher.
398- // its adapter receive-end mark before cmd.Wait completes.415+ order = append(order, "descendant_receive_end")
399- order = append(order, "adapter_receive_end")416+ waitErr <- nil
400- waitErr <- nil
401- },
402- close: func() { order = append(order, "pty_closed") },
403 },417 },
404- output: &orderedOutput{close: func() { order = append(order, "output_closed") }},
405- waitErr: waitErr,
406- waitTimeout: func() <-chan time.Time { return timeout },
407 forceCleanup: func() {418 forceCleanup: func() {
408- order = append(order, "forced_process_group_cleanup")419+ t.Fatal("normal shutdown sent SIGTERM to the traced descendant process group")
409 },420 },
410 }421 }
411 422
412 if err := p.Close(); err != nil {423 if err := p.Close(); err != nil {
413 t.Fatal(err)424 t.Fatal(err)
414 }425 }
415- if !equalStrings(order, []string{"exit_written", "adapter_receive_end", "pty_closed", "output_closed"}) {426+ if !equalStrings(order, []string{"shell_exit", "descendant_receive_end", "pty_closed", "output_closed"}) {
416 t.Fatalf("cleanup order=%q", order)427 t.Fatalf("cleanup order=%q", order)
417 }428 }
418 }429 }
M internal/adapters/snapshot/dirent_cookie_darwin.go +6 -2
@@ -14,8 +14,12 @@ import (
1414 // descriptor-maintained count is saved.
1515 const maintenanceDirentBufferSize = int(unsafe.Sizeof(syscall.Dirent{}))
1616
17-func directoryCookie(file maintenanceDirectory, _ *syscall.Dirent) (int64, error) {
18- return file.Seek(0, io.SeekCurrent)
17+func directoryCookie(file maintenanceDirectory, _ *syscall.Dirent, remaining int) (int64, error) {
18+ end, err := file.Seek(0, io.SeekCurrent)
19+ if err != nil {
20+ return 0, err
21+ }
22+ return end - int64(remaining), nil
1923 }
2024
2125 func drainMaintenanceDirentBatch() bool { return true }
@@ -14,8 +14,12 @@ import (
14 // descriptor-maintained count is saved.14 // descriptor-maintained count is saved.
15 const maintenanceDirentBufferSize = int(unsafe.Sizeof(syscall.Dirent{}))15 const maintenanceDirentBufferSize = int(unsafe.Sizeof(syscall.Dirent{}))
16 16
17-func directoryCookie(file maintenanceDirectory, _ *syscall.Dirent) (int64, error) {17+func directoryCookie(file maintenanceDirectory, _ *syscall.Dirent, remaining int) (int64, error) {
18- return file.Seek(0, io.SeekCurrent)18+ end, err := file.Seek(0, io.SeekCurrent)
19+ if err != nil {
20+ return 0, err
21+ }
22+ return end - int64(remaining), nil
19 }23 }
20 24
21 func drainMaintenanceDirentBatch() bool { return true }25 func drainMaintenanceDirentBatch() bool { return true }
M internal/adapters/snapshot/dirent_cookie_darwin_test.go +12 -1
@@ -10,7 +10,7 @@ import (
1010 func TestDirectoryCookieUsesDescriptorEntryCount(t *testing.T) {
1111 file := fakeMaintenanceDirectory{seekOffset: 42}
1212 dirent := syscall.Dirent{Seekoff: 99}
13- got, err := directoryCookie(file, &dirent)
13+ got, err := directoryCookie(file, &dirent, 0)
1414 if err != nil {
1515 t.Fatal(err)
1616 }
@@ -18,3 +18,14 @@ func TestDirectoryCookieUsesDescriptorEntryCount(t *testing.T) {
1818 t.Fatalf("directoryCookie() = %d, want descriptor entry count 42", got)
1919 }
2020 }
21+
22+func TestDirectoryCookieSubtractsBufferedEntries(t *testing.T) {
23+ file := fakeMaintenanceDirectory{seekOffset: 42}
24+ got, err := directoryCookie(file, &syscall.Dirent{}, 3)
25+ if err != nil {
26+ t.Fatal(err)
27+ }
28+ if got != 39 {
29+ t.Fatalf("directoryCookie() = %d, want per-entry cursor 39", got)
30+ }
31+}
@@ -10,7 +10,7 @@ import (
10 func TestDirectoryCookieUsesDescriptorEntryCount(t *testing.T) {10 func TestDirectoryCookieUsesDescriptorEntryCount(t *testing.T) {
11 file := fakeMaintenanceDirectory{seekOffset: 42}11 file := fakeMaintenanceDirectory{seekOffset: 42}
12 dirent := syscall.Dirent{Seekoff: 99}12 dirent := syscall.Dirent{Seekoff: 99}
13- got, err := directoryCookie(file, &dirent)13+ got, err := directoryCookie(file, &dirent, 0)
14 if err != nil {14 if err != nil {
15 t.Fatal(err)15 t.Fatal(err)
16 }16 }
@@ -18,3 +18,14 @@ func TestDirectoryCookieUsesDescriptorEntryCount(t *testing.T) {
18 t.Fatalf("directoryCookie() = %d, want descriptor entry count 42", got)18 t.Fatalf("directoryCookie() = %d, want descriptor entry count 42", got)
19 }19 }
20 }20 }
21+
22+func TestDirectoryCookieSubtractsBufferedEntries(t *testing.T) {
23+ file := fakeMaintenanceDirectory{seekOffset: 42}
24+ got, err := directoryCookie(file, &syscall.Dirent{}, 3)
25+ if err != nil {
26+ t.Fatal(err)
27+ }
28+ if got != 39 {
29+ t.Fatalf("directoryCookie() = %d, want per-entry cursor 39", got)
30+ }
31+}
M internal/adapters/snapshot/dirent_cookie_linux.go +1 -1
@@ -7,7 +7,7 @@ import "syscall"
77 const maintenanceDirentBufferSize = 8192
88
99 // directoryCookie is d_off, the resumable getdents(2) cookie for Linux.
10-func directoryCookie(_ maintenanceDirectory, dirent *syscall.Dirent) (int64, error) {
10+func directoryCookie(_ maintenanceDirectory, dirent *syscall.Dirent, _ int) (int64, error) {
1111 return dirent.Off, nil
1212 }
1313
@@ -7,7 +7,7 @@ import "syscall"
7 const maintenanceDirentBufferSize = 81927 const maintenanceDirentBufferSize = 8192
8 8
9 // directoryCookie is d_off, the resumable getdents(2) cookie for Linux.9 // directoryCookie is d_off, the resumable getdents(2) cookie for Linux.
10-func directoryCookie(_ maintenanceDirectory, dirent *syscall.Dirent) (int64, error) {10+func directoryCookie(_ maintenanceDirectory, dirent *syscall.Dirent, _ int) (int64, error) {
11 return dirent.Off, nil11 return dirent.Off, nil
12 }12 }
13 13
M internal/adapters/snapshot/dirent_cookie_linux_test.go +1 -1
@@ -9,7 +9,7 @@ import (
99
1010 func TestDirectoryCookieUsesLinuxOffset(t *testing.T) {
1111 dirent := syscall.Dirent{Off: 42}
12- got, err := directoryCookie(fakeMaintenanceDirectory{}, &dirent)
12+ got, err := directoryCookie(fakeMaintenanceDirectory{}, &dirent, 0)
1313 if err != nil {
1414 t.Fatal(err)
1515 }
@@ -9,7 +9,7 @@ import (
9 9
10 func TestDirectoryCookieUsesLinuxOffset(t *testing.T) {10 func TestDirectoryCookieUsesLinuxOffset(t *testing.T) {
11 dirent := syscall.Dirent{Off: 42}11 dirent := syscall.Dirent{Off: 42}
12- got, err := directoryCookie(fakeMaintenanceDirectory{}, &dirent)12+ got, err := directoryCookie(fakeMaintenanceDirectory{}, &dirent, 0)
13 if err != nil {13 if err != nil {
14 t.Fatal(err)14 t.Fatal(err)
15 }15 }
M internal/adapters/snapshot/maintenance_cursor.go +46 -18
@@ -26,8 +26,14 @@ type maintenanceCursor struct {
2626 }
2727
2828 type maintenanceDirEntry struct {
29- name string
30- isDir bool
29+ name string
30+ isDir bool
31+ cookie int64
32+}
33+
34+type maintenanceDirentConfig struct {
35+ drainBuffer bool
36+ cookie func(maintenanceDirectory, *syscall.Dirent, int) (int64, error)
3137 }
3238
3339 func (r *Repository) resetMaintenance() {
@@ -49,16 +55,6 @@ func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" +
4955 func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]maintenanceDirEntry, bool, error) {
5056 id := maintenanceCursorID(dir, purpose)
5157 cursor := r.maintenanceCursors[id]
52- if cursor != nil && len(cursor.pending) != 0 {
53- count := min(n, len(cursor.pending))
54- entries := cursor.pending[:count]
55- cursor.pending = cursor.pending[count:]
56- done := cursor.done && len(cursor.pending) == 0
57- if done {
58- delete(r.maintenanceCursors, id)
59- }
60- return entries, done, nil
61- }
6258 if cursor == nil {
6359 if len(r.maintenanceCursors) >= maxMaintenanceCursors {
6460 return nil, false, errMaintenanceCursorLimit
@@ -121,13 +117,26 @@ func maintenanceDirectoryError(operation string, err error) error {
121117 // stores an entry count in the descriptor offset, so every returned buffer
122118 // must be drained before that descriptor-maintained count is saved.
123119 func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) (entries []maintenanceDirEntry, done bool, err error) {
124- return r.readMaintenanceDirentWithDrain(dir, limit, cursor, drainMaintenanceDirentBatch(), directoryCookie)
120+ if len(cursor.pending) != 0 {
121+ count := min(limit, len(cursor.pending))
122+ entries = cursor.pending[:count]
123+ cursor.pending = cursor.pending[count:]
124+ if len(entries) != 0 {
125+ cursor.offset = entries[len(entries)-1].cookie
126+ }
127+ return entries, cursor.done && len(cursor.pending) == 0, nil
128+ }
129+ config := maintenanceDirentConfig{drainBuffer: drainMaintenanceDirentBatch(), cookie: directoryCookie}
130+ if r.hooks.maintenanceDirentConfig != nil {
131+ config = *r.hooks.maintenanceDirentConfig
132+ }
133+ return r.readMaintenanceDirentWithDrain(dir, limit, cursor, config.drainBuffer, config.cookie)
125134 }
126135
127136 // readMaintenanceDirentWithDrain contains the shared traversal logic. The
128137 // explicit parameters permit platform-independent verification of Darwin's
129138 // buffer-draining semantics with deterministic fake directories.
130-func (r *Repository) readMaintenanceDirentWithDrain(dir string, limit int, cursor *maintenanceCursor, drainBuffer bool, cookie func(maintenanceDirectory, *syscall.Dirent) (int64, error)) (entries []maintenanceDirEntry, done bool, err error) {
139+func (r *Repository) readMaintenanceDirentWithDrain(dir string, limit int, cursor *maintenanceCursor, drainBuffer bool, cookie func(maintenanceDirectory, *syscall.Dirent, int) (int64, error)) (entries []maintenanceDirEntry, done bool, err error) {
131140 file, err := r.openMaintenanceDirectory(dir)
132141 if err != nil {
133142 return nil, false, maintenanceDirectoryError("open maintenance directory", err)
@@ -163,14 +172,19 @@ func (r *Repository) readMaintenanceDirentWithDrain(dir string, limit int, curso
163172 if count == 0 {
164173 return entries, true, nil
165174 }
175+ remainingRecords, err := countMaintenanceDirents(buffer[:count])
176+ if err != nil {
177+ return nil, false, err
178+ }
166179 for data := buffer[:count]; len(data) != 0 && (len(entries) < limit || drainBuffer); {
167180 record, nameBytes, reclen, decodeErr := decodeMaintenanceDirent(data)
168181 if decodeErr != nil {
169182 return nil, false, decodeErr
170183 }
171184 data = data[reclen:]
185+ remainingRecords--
172186 // Advance past every raw record, including dot and disappeared entries.
173- offset, cookieErr := cookie(file, record)
187+ offset, cookieErr := cookie(file, record, remainingRecords)
174188 if cookieErr != nil {
175189 return nil, false, maintenanceDirectoryError("tell maintenance directory", cookieErr)
176190 }
@@ -189,19 +203,33 @@ func (r *Repository) readMaintenanceDirentWithDrain(dir string, limit int, curso
189203 if err != nil {
190204 return nil, false, maintenanceDirectoryError("stat maintenance directory entry", err)
191205 }
192- entries = append(entries, maintenanceDirEntry{name: name, isDir: stat.IsDir()})
206+ entries = append(entries, maintenanceDirEntry{name: name, isDir: stat.IsDir(), cookie: offset})
193207 }
194208 if len(entries) >= limit {
195209 if drainBuffer && len(entries) > limit {
196- // Darwin has already advanced the descriptor past this entire
197- // buffer, so retain only this final buffer's excess.
210+ // The descriptor is past the whole buffer. Retain the excess and
211+ // expose only the last entry actually consumed as the logical cursor.
198212 cursor.pending = append(cursor.pending, entries[limit:]...)
213+ cursor.offset = entries[limit-1].cookie
199214 }
200215 return entries[:limit], false, nil
201216 }
202217 }
203218 }
204219
220+func countMaintenanceDirents(data []byte) (int, error) {
221+ count := 0
222+ for len(data) != 0 {
223+ _, _, reclen, err := decodeMaintenanceDirent(data)
224+ if err != nil {
225+ return 0, err
226+ }
227+ data = data[reclen:]
228+ count++
229+ }
230+ return count, nil
231+}
232+
205233 // decodeMaintenanceDirent copies a bounded raw record into aligned storage
206234 // before reading its fields. syscall.ReadDirent writes to a byte buffer, whose
207235 // start and record boundaries are not guaranteed to satisfy syscall.Dirent's
@@ -26,8 +26,14 @@ type maintenanceCursor struct {
26 }26 }
27 27
28 type maintenanceDirEntry struct {28 type maintenanceDirEntry struct {
29- name string29+ name string
30- isDir bool30+ isDir bool
31+ cookie int64
32+}
33+
34+type maintenanceDirentConfig struct {
35+ drainBuffer bool
36+ cookie func(maintenanceDirectory, *syscall.Dirent, int) (int64, error)
31 }37 }
32 38
33 func (r *Repository) resetMaintenance() {39 func (r *Repository) resetMaintenance() {
@@ -49,16 +55,6 @@ func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" +
49 func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]maintenanceDirEntry, bool, error) {55 func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]maintenanceDirEntry, bool, error) {
50 id := maintenanceCursorID(dir, purpose)56 id := maintenanceCursorID(dir, purpose)
51 cursor := r.maintenanceCursors[id]57 cursor := r.maintenanceCursors[id]
52- if cursor != nil && len(cursor.pending) != 0 {
53- count := min(n, len(cursor.pending))
54- entries := cursor.pending[:count]
55- cursor.pending = cursor.pending[count:]
56- done := cursor.done && len(cursor.pending) == 0
57- if done {
58- delete(r.maintenanceCursors, id)
59- }
60- return entries, done, nil
61- }
62 if cursor == nil {58 if cursor == nil {
63 if len(r.maintenanceCursors) >= maxMaintenanceCursors {59 if len(r.maintenanceCursors) >= maxMaintenanceCursors {
64 return nil, false, errMaintenanceCursorLimit60 return nil, false, errMaintenanceCursorLimit
@@ -121,13 +117,26 @@ func maintenanceDirectoryError(operation string, err error) error {
121 // stores an entry count in the descriptor offset, so every returned buffer117 // stores an entry count in the descriptor offset, so every returned buffer
122 // must be drained before that descriptor-maintained count is saved.118 // must be drained before that descriptor-maintained count is saved.
123 func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) (entries []maintenanceDirEntry, done bool, err error) {119 func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) (entries []maintenanceDirEntry, done bool, err error) {
124- return r.readMaintenanceDirentWithDrain(dir, limit, cursor, drainMaintenanceDirentBatch(), directoryCookie)120+ if len(cursor.pending) != 0 {
121+ count := min(limit, len(cursor.pending))
122+ entries = cursor.pending[:count]
123+ cursor.pending = cursor.pending[count:]
124+ if len(entries) != 0 {
125+ cursor.offset = entries[len(entries)-1].cookie
126+ }
127+ return entries, cursor.done && len(cursor.pending) == 0, nil
128+ }
129+ config := maintenanceDirentConfig{drainBuffer: drainMaintenanceDirentBatch(), cookie: directoryCookie}
130+ if r.hooks.maintenanceDirentConfig != nil {
131+ config = *r.hooks.maintenanceDirentConfig
132+ }
133+ return r.readMaintenanceDirentWithDrain(dir, limit, cursor, config.drainBuffer, config.cookie)
125 }134 }
126 135
127 // readMaintenanceDirentWithDrain contains the shared traversal logic. The136 // readMaintenanceDirentWithDrain contains the shared traversal logic. The
128 // explicit parameters permit platform-independent verification of Darwin's137 // explicit parameters permit platform-independent verification of Darwin's
129 // buffer-draining semantics with deterministic fake directories.138 // buffer-draining semantics with deterministic fake directories.
130-func (r *Repository) readMaintenanceDirentWithDrain(dir string, limit int, cursor *maintenanceCursor, drainBuffer bool, cookie func(maintenanceDirectory, *syscall.Dirent) (int64, error)) (entries []maintenanceDirEntry, done bool, err error) {139+func (r *Repository) readMaintenanceDirentWithDrain(dir string, limit int, cursor *maintenanceCursor, drainBuffer bool, cookie func(maintenanceDirectory, *syscall.Dirent, int) (int64, error)) (entries []maintenanceDirEntry, done bool, err error) {
131 file, err := r.openMaintenanceDirectory(dir)140 file, err := r.openMaintenanceDirectory(dir)
132 if err != nil {141 if err != nil {
133 return nil, false, maintenanceDirectoryError("open maintenance directory", err)142 return nil, false, maintenanceDirectoryError("open maintenance directory", err)
@@ -163,14 +172,19 @@ func (r *Repository) readMaintenanceDirentWithDrain(dir string, limit int, curso
163 if count == 0 {172 if count == 0 {
164 return entries, true, nil173 return entries, true, nil
165 }174 }
175+ remainingRecords, err := countMaintenanceDirents(buffer[:count])
176+ if err != nil {
177+ return nil, false, err
178+ }
166 for data := buffer[:count]; len(data) != 0 && (len(entries) < limit || drainBuffer); {179 for data := buffer[:count]; len(data) != 0 && (len(entries) < limit || drainBuffer); {
167 record, nameBytes, reclen, decodeErr := decodeMaintenanceDirent(data)180 record, nameBytes, reclen, decodeErr := decodeMaintenanceDirent(data)
168 if decodeErr != nil {181 if decodeErr != nil {
169 return nil, false, decodeErr182 return nil, false, decodeErr
170 }183 }
171 data = data[reclen:]184 data = data[reclen:]
185+ remainingRecords--
172 // Advance past every raw record, including dot and disappeared entries.186 // Advance past every raw record, including dot and disappeared entries.
173- offset, cookieErr := cookie(file, record)187+ offset, cookieErr := cookie(file, record, remainingRecords)
174 if cookieErr != nil {188 if cookieErr != nil {
175 return nil, false, maintenanceDirectoryError("tell maintenance directory", cookieErr)189 return nil, false, maintenanceDirectoryError("tell maintenance directory", cookieErr)
176 }190 }
@@ -189,19 +203,33 @@ func (r *Repository) readMaintenanceDirentWithDrain(dir string, limit int, curso
189 if err != nil {203 if err != nil {
190 return nil, false, maintenanceDirectoryError("stat maintenance directory entry", err)204 return nil, false, maintenanceDirectoryError("stat maintenance directory entry", err)
191 }205 }
192- entries = append(entries, maintenanceDirEntry{name: name, isDir: stat.IsDir()})206+ entries = append(entries, maintenanceDirEntry{name: name, isDir: stat.IsDir(), cookie: offset})
193 }207 }
194 if len(entries) >= limit {208 if len(entries) >= limit {
195 if drainBuffer && len(entries) > limit {209 if drainBuffer && len(entries) > limit {
196- // Darwin has already advanced the descriptor past this entire210+ // The descriptor is past the whole buffer. Retain the excess and
197- // buffer, so retain only this final buffer's excess.211+ // expose only the last entry actually consumed as the logical cursor.
198 cursor.pending = append(cursor.pending, entries[limit:]...)212 cursor.pending = append(cursor.pending, entries[limit:]...)
213+ cursor.offset = entries[limit-1].cookie
199 }214 }
200 return entries[:limit], false, nil215 return entries[:limit], false, nil
201 }216 }
202 }217 }
203 }218 }
204 219
220+func countMaintenanceDirents(data []byte) (int, error) {
221+ count := 0
222+ for len(data) != 0 {
223+ _, _, reclen, err := decodeMaintenanceDirent(data)
224+ if err != nil {
225+ return 0, err
226+ }
227+ data = data[reclen:]
228+ count++
229+ }
230+ return count, nil
231+}
232+
205 // decodeMaintenanceDirent copies a bounded raw record into aligned storage233 // decodeMaintenanceDirent copies a bounded raw record into aligned storage
206 // before reading its fields. syscall.ReadDirent writes to a byte buffer, whose234 // before reading its fields. syscall.ReadDirent writes to a byte buffer, whose
207 // start and record boundaries are not guaranteed to satisfy syscall.Dirent's235 // start and record boundaries are not guaranteed to satisfy syscall.Dirent's
M internal/adapters/snapshot/maintenance_gc.go +23 -5
@@ -630,6 +630,9 @@ func (r *Repository) readRetentionDir(state *retentionMaintenance, dir string, n
630630 count := min(n, len(cursor.pending))
631631 entries := cursor.pending[:count]
632632 cursor.pending = cursor.pending[count:]
633+ if len(entries) != 0 {
634+ cursor.offset = entries[len(entries)-1].cookie
635+ }
633636 done := cursor.done && len(cursor.pending) == 0
634637 if done {
635638 delete(state.cursors, id)
@@ -787,6 +790,7 @@ func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRe
787790 }
788791 findings = make([]ports.ReconcileFinding, 0, findingCapacity)
789792 stepConsumed := ports.MaintenanceBudget{}
793+ resumeCursor := cursor
790794 for scanned := uint64(0); scanned < budget.Entries && stepConsumed.Entries < budget.Entries; scanned++ {
791795 entries, done, err := r.readMaintenanceDirent(root, 1, state)
792796 if errors.Is(err, os.ErrNotExist) {
@@ -802,34 +806,48 @@ func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRe
802806 continue
803807 }
804808 entry := entries[0]
805- next := ports.ReconcileCursor{DirectoryCookie: uint64(state.offset)}
809+ nextCursor = ports.ReconcileCursor{DirectoryCookie: uint64(entry.cookie)}
806810 if !entry.isDir || isQuarantine(entry.name) {
811+ resumeCursor = nextCursor
807812 if done {
808813 return ports.ReconcileCursor{}, findings, nil
809814 }
810815 continue
811816 }
812817 findingIndex := len(findings)
818+ consumedBeforeCandidate := stepConsumed
813819 var id domain.IncarnationID
814820 if err := id.UnmarshalText([]byte(entry.name)); err != nil {
815- findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name}, Cursor: next, Consumed: ports.MaintenanceBudget{Entries: 1}})
821+ findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name}, Cursor: nextCursor, Consumed: ports.MaintenanceBudget{Entries: 1}})
816822 } else if record, ok := byIncarnation[id]; !ok {
817- findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name, IncarnationID: id}, Cursor: next, Consumed: ports.MaintenanceBudget{Entries: 1}})
823+ findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name, IncarnationID: id}, Cursor: nextCursor, Consumed: ports.MaintenanceBudget{Entries: 1}})
818824 } else {
819825 remaining := ports.MaintenanceBudget{Entries: budget.Entries - stepConsumed.Entries, Bytes: budget.Bytes - stepConsumed.Bytes}
820826 finding := r.reconcileIncarnation(ctx, record, remaining)
821- finding.Cursor = next
827+ finding.Cursor = nextCursor
822828 findings = append(findings, finding)
823829 }
824830 if len(findings) > findingIndex {
825831 stepConsumed.Entries += findings[findingIndex].Consumed.Entries
826832 stepConsumed.Bytes += findings[findingIndex].Consumed.Bytes
833+ if findings[findingIndex].Status == ports.ReconcileBudgetExhausted {
834+ // If a fresh per-call budget cannot admit this candidate, retrying
835+ // the same fixed budget can never succeed. Report the bounded
836+ // exhaustion once and deliberately advance so later entries remain
837+ // reachable. A candidate reached after earlier work is retried on a
838+ // fresh call because it may fit that call's full budget.
839+ if consumedBeforeCandidate.Entries == 0 && consumedBeforeCandidate.Bytes == 0 {
840+ return nextCursor, findings, nil
841+ }
842+ return resumeCursor, findings, nil
843+ }
827844 }
845+ resumeCursor = nextCursor
828846 if done {
829847 return ports.ReconcileCursor{}, findings, nil
830848 }
831849 }
832- return ports.ReconcileCursor{DirectoryCookie: uint64(state.offset)}, findings, nil
850+ return nextCursor, findings, nil
833851 }
834852
835853 func (r *Repository) reconcileIncarnation(ctx context.Context, record domain.CatalogueRecord, budget ports.MaintenanceBudget) ports.ReconcileFinding {
@@ -630,6 +630,9 @@ func (r *Repository) readRetentionDir(state *retentionMaintenance, dir string, n
630 count := min(n, len(cursor.pending))630 count := min(n, len(cursor.pending))
631 entries := cursor.pending[:count]631 entries := cursor.pending[:count]
632 cursor.pending = cursor.pending[count:]632 cursor.pending = cursor.pending[count:]
633+ if len(entries) != 0 {
634+ cursor.offset = entries[len(entries)-1].cookie
635+ }
633 done := cursor.done && len(cursor.pending) == 0636 done := cursor.done && len(cursor.pending) == 0
634 if done {637 if done {
635 delete(state.cursors, id)638 delete(state.cursors, id)
@@ -787,6 +790,7 @@ func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRe
787 }790 }
788 findings = make([]ports.ReconcileFinding, 0, findingCapacity)791 findings = make([]ports.ReconcileFinding, 0, findingCapacity)
789 stepConsumed := ports.MaintenanceBudget{}792 stepConsumed := ports.MaintenanceBudget{}
793+ resumeCursor := cursor
790 for scanned := uint64(0); scanned < budget.Entries && stepConsumed.Entries < budget.Entries; scanned++ {794 for scanned := uint64(0); scanned < budget.Entries && stepConsumed.Entries < budget.Entries; scanned++ {
791 entries, done, err := r.readMaintenanceDirent(root, 1, state)795 entries, done, err := r.readMaintenanceDirent(root, 1, state)
792 if errors.Is(err, os.ErrNotExist) {796 if errors.Is(err, os.ErrNotExist) {
@@ -802,34 +806,48 @@ func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRe
802 continue806 continue
803 }807 }
804 entry := entries[0]808 entry := entries[0]
805- next := ports.ReconcileCursor{DirectoryCookie: uint64(state.offset)}809+ nextCursor = ports.ReconcileCursor{DirectoryCookie: uint64(entry.cookie)}
806 if !entry.isDir || isQuarantine(entry.name) {810 if !entry.isDir || isQuarantine(entry.name) {
811+ resumeCursor = nextCursor
807 if done {812 if done {
808 return ports.ReconcileCursor{}, findings, nil813 return ports.ReconcileCursor{}, findings, nil
809 }814 }
810 continue815 continue
811 }816 }
812 findingIndex := len(findings)817 findingIndex := len(findings)
818+ consumedBeforeCandidate := stepConsumed
813 var id domain.IncarnationID819 var id domain.IncarnationID
814 if err := id.UnmarshalText([]byte(entry.name)); err != nil {820 if err := id.UnmarshalText([]byte(entry.name)); err != nil {
815- findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name}, Cursor: next, Consumed: ports.MaintenanceBudget{Entries: 1}})821+ findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name}, Cursor: nextCursor, Consumed: ports.MaintenanceBudget{Entries: 1}})
816 } else if record, ok := byIncarnation[id]; !ok {822 } else if record, ok := byIncarnation[id]; !ok {
817- findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name, IncarnationID: id}, Cursor: next, Consumed: ports.MaintenanceBudget{Entries: 1}})823+ findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name, IncarnationID: id}, Cursor: nextCursor, Consumed: ports.MaintenanceBudget{Entries: 1}})
818 } else {824 } else {
819 remaining := ports.MaintenanceBudget{Entries: budget.Entries - stepConsumed.Entries, Bytes: budget.Bytes - stepConsumed.Bytes}825 remaining := ports.MaintenanceBudget{Entries: budget.Entries - stepConsumed.Entries, Bytes: budget.Bytes - stepConsumed.Bytes}
820 finding := r.reconcileIncarnation(ctx, record, remaining)826 finding := r.reconcileIncarnation(ctx, record, remaining)
821- finding.Cursor = next827+ finding.Cursor = nextCursor
822 findings = append(findings, finding)828 findings = append(findings, finding)
823 }829 }
824 if len(findings) > findingIndex {830 if len(findings) > findingIndex {
825 stepConsumed.Entries += findings[findingIndex].Consumed.Entries831 stepConsumed.Entries += findings[findingIndex].Consumed.Entries
826 stepConsumed.Bytes += findings[findingIndex].Consumed.Bytes832 stepConsumed.Bytes += findings[findingIndex].Consumed.Bytes
833+ if findings[findingIndex].Status == ports.ReconcileBudgetExhausted {
834+ // If a fresh per-call budget cannot admit this candidate, retrying
835+ // the same fixed budget can never succeed. Report the bounded
836+ // exhaustion once and deliberately advance so later entries remain
837+ // reachable. A candidate reached after earlier work is retried on a
838+ // fresh call because it may fit that call's full budget.
839+ if consumedBeforeCandidate.Entries == 0 && consumedBeforeCandidate.Bytes == 0 {
840+ return nextCursor, findings, nil
841+ }
842+ return resumeCursor, findings, nil
843+ }
827 }844 }
845+ resumeCursor = nextCursor
828 if done {846 if done {
829 return ports.ReconcileCursor{}, findings, nil847 return ports.ReconcileCursor{}, findings, nil
830 }848 }
831 }849 }
832- return ports.ReconcileCursor{DirectoryCookie: uint64(state.offset)}, findings, nil850+ return nextCursor, findings, nil
833 }851 }
834 852
835 func (r *Repository) reconcileIncarnation(ctx context.Context, record domain.CatalogueRecord, budget ports.MaintenanceBudget) ports.ReconcileFinding {853 func (r *Repository) reconcileIncarnation(ctx context.Context, record domain.CatalogueRecord, budget ports.MaintenanceBudget) ports.ReconcileFinding {
M internal/adapters/snapshot/maintenance_test.go +1 -1
@@ -287,7 +287,7 @@ func (d *fakeMultiBatchMaintenanceDirectory) ReadDirent(buffer []byte) (int, err
287287
288288 func (d *fakeMultiBatchMaintenanceDirectory) Close() error { return nil }
289289
290-func maintenanceTestDirectoryCookie(file maintenanceDirectory, _ *syscall.Dirent) (int64, error) {
290+func maintenanceTestDirectoryCookie(file maintenanceDirectory, _ *syscall.Dirent, _ int) (int64, error) {
291291 return file.Seek(0, io.SeekCurrent)
292292 }
293293
@@ -287,7 +287,7 @@ func (d *fakeMultiBatchMaintenanceDirectory) ReadDirent(buffer []byte) (int, err
287 287
288 func (d *fakeMultiBatchMaintenanceDirectory) Close() error { return nil }288 func (d *fakeMultiBatchMaintenanceDirectory) Close() error { return nil }
289 289
290-func maintenanceTestDirectoryCookie(file maintenanceDirectory, _ *syscall.Dirent) (int64, error) {290+func maintenanceTestDirectoryCookie(file maintenanceDirectory, _ *syscall.Dirent, _ int) (int64, error) {
291 return file.Seek(0, io.SeekCurrent)291 return file.Seek(0, io.SeekCurrent)
292 }292 }
293 293
M internal/adapters/snapshot/reconciliation_test.go +122 -4
@@ -2,7 +2,9 @@ package snapshot
22
33 import (
44 "context"
5+ "io"
56 "path/filepath"
7+ "syscall"
68 "testing"
79
810 "github.com/bnema/vev/internal/domain"
@@ -12,6 +14,72 @@ import (
1214 "github.com/stretchr/testify/require"
1315 )
1416
17+type fakeReconcileDrainDirectory struct {
18+ names []string
19+ position int
20+}
21+
22+func (d *fakeReconcileDrainDirectory) Seek(offset int64, whence int) (int64, error) {
23+ if whence == io.SeekCurrent {
24+ return int64(d.position), nil
25+ }
26+ d.position = int(offset)
27+ return offset, nil
28+}
29+
30+func (d *fakeReconcileDrainDirectory) ReadDirent(buffer []byte) (int, error) {
31+ if d.position >= len(d.names) {
32+ return 0, nil
33+ }
34+ batch := maintenanceDirentBatch(d.names[d.position:]...)
35+ d.position = len(d.names)
36+ return copy(buffer, batch), nil
37+}
38+
39+func (d *fakeReconcileDrainDirectory) Close() error { return nil }
40+
41+func configureDrainingReconcileScanner(repo *Repository, names []string) {
42+ repo.hooks.openMaintenanceDirectory = func(string) (maintenanceDirectory, error) {
43+ return &fakeReconcileDrainDirectory{names: names}, nil
44+ }
45+ repo.hooks.maintenanceDirentConfig = &maintenanceDirentConfig{
46+ drainBuffer: true,
47+ cookie: func(file maintenanceDirectory, _ *syscall.Dirent, remaining int) (int64, error) {
48+ end, err := file.Seek(0, io.SeekCurrent)
49+ return end - int64(remaining), err
50+ },
51+ }
52+}
53+
54+func TestReconciliationDrainedBatchHasNoSkippedOrDuplicateEntries(t *testing.T) {
55+ repo := NewRepository(privateDir(t))
56+ sessions := filepath.Join(repo.dir, repositorySessionsDir)
57+ names := make([]string, 3)
58+ for i := range names {
59+ id := domain.IncarnationID{byte(i + 20)}
60+ names[i] = id.String()
61+ require.NoError(t, safedir.EnsurePrivate(filepath.Join(sessions, names[i])))
62+ }
63+ configureDrainingReconcileScanner(repo, names)
64+
65+ cursor := ports.ReconcileCursor{}
66+ var got []string
67+ for calls := 0; ; calls++ {
68+ next, findings, err := repo.Reconcile(context.Background(), nil, cursor, ports.MaintenanceBudget{Entries: 2, Bytes: 1})
69+ require.NoError(t, err)
70+ for _, finding := range findings {
71+ got = append(got, finding.Candidate.Name)
72+ }
73+ if next.DirectoryCookie == 0 {
74+ break
75+ }
76+ require.Greater(t, next.DirectoryCookie, cursor.DirectoryCookie)
77+ cursor = next
78+ require.Less(t, calls, 3)
79+ }
80+ require.Equal(t, names, got)
81+}
82+
1583 func TestReconciliationReadsEachCandidatePayloadOnceWithinBudget(t *testing.T) {
1684 repo := NewRepository(privateDir(t))
1785 first := publishMaintenanceGenerations(t, repo, "first", 2)
@@ -94,6 +162,50 @@ func TestReconciliationReadsEachCandidatePayloadOnceWithinBudget(t *testing.T) {
94162 }
95163 }
96164
165+func TestReconciliationIntrinsicallyOversizedCandidateMakesFixedBudgetProgress(t *testing.T) {
166+ tests := []struct {
167+ name string
168+ budget ports.MaintenanceBudget
169+ }{
170+ {name: "entry cost", budget: ports.MaintenanceBudget{Entries: 1, Bytes: 8 << 20}},
171+ {name: "byte cost", budget: ports.MaintenanceBudget{Entries: 8, Bytes: 1}},
172+ }
173+ for _, tc := range tests {
174+ t.Run(tc.name, func(t *testing.T) {
175+ repo := NewRepository(privateDir(t))
176+ publications := publishMaintenanceGenerations(t, repo, "work", 2)
177+ committed := checkpointRefForPublication(publications[0])
178+ record := domain.CatalogueRecord{Name: "work", IncarnationID: publications[0].IncarnationID, RecoveryState: domain.RecoveryHealthy, Committed: &committed}
179+ unknown := domain.IncarnationID{99}
180+ require.NoError(t, safedir.EnsurePrivate(filepath.Join(repo.dir, repositorySessionsDir, unknown.String())))
181+ configureDrainingReconcileScanner(repo, []string{record.IncarnationID.String(), unknown.String()})
182+
183+ next, findings, err := repo.Reconcile(context.Background(), []domain.CatalogueRecord{record}, ports.ReconcileCursor{}, tc.budget)
184+ require.NoError(t, err)
185+ require.Equal(t, uint64(1), next.DirectoryCookie)
186+ require.Len(t, findings, 1)
187+ require.Equal(t, ports.ReconcileBudgetExhausted, findings[0].Status)
188+ require.LessOrEqual(t, findings[0].Consumed.Entries, tc.budget.Entries)
189+ require.LessOrEqual(t, findings[0].Consumed.Bytes, tc.budget.Bytes)
190+
191+ later, findings, err := repo.Reconcile(context.Background(), []domain.CatalogueRecord{record}, next, tc.budget)
192+ require.NoError(t, err)
193+ require.Len(t, findings, 1)
194+ require.Equal(t, unknown, findings[0].Candidate.IncarnationID)
195+ require.Equal(t, ports.ReconcileQuarantined, findings[0].Status)
196+ if later.DirectoryCookie == 0 {
197+ return
198+ }
199+ require.Greater(t, later.DirectoryCookie, next.DirectoryCookie)
200+
201+ done, findings, err := repo.Reconcile(context.Background(), []domain.CatalogueRecord{record}, later, tc.budget)
202+ require.NoError(t, err)
203+ require.Zero(t, done)
204+ require.Empty(t, findings, "oversized candidate must not be retried")
205+ })
206+ }
207+}
208+
97209 func TestReconciliationCursor(t *testing.T) {
98210 repo := NewRepository(privateDir(t))
99211 publications := publishMaintenanceGenerations(t, repo, "work", 2)
@@ -132,19 +244,25 @@ func TestReconciliationCursor(t *testing.T) {
132244 }
133245 require.Greater(t, calls, 1, "top-level traversal must resume across bounded calls")
134246
135- unknown := 0
247+ unknown := make(map[domain.IncarnationID]struct{})
136248 adoptable := 0
137249 for _, finding := range findings {
138250 switch finding.Kind {
139251 case ports.ReconcileUnknownIncarnation:
140- unknown++
252+ require.Equal(t, ports.ReconcileQuarantined, finding.Status)
253+ _, duplicate := unknown[finding.Candidate.IncarnationID]
254+ require.False(t, duplicate, "unknown incarnation was returned twice")
255+ unknown[finding.Candidate.IncarnationID] = struct{}{}
141256 case ports.ReconcileForwardOrphan:
257+ if finding.Status == ports.ReconcileBudgetExhausted {
258+ continue
259+ }
142260 require.Equal(t, ports.ReconcileValidated, finding.Status)
143261 require.Equal(t, uint64(2), finding.Candidate.Ref.Generation)
144262 require.Equal(t, committed, *finding.Candidate.Parent)
145263 adoptable++
146264 }
147265 }
148- require.Equal(t, 6, unknown)
149- require.Equal(t, 1, adoptable)
266+ require.Len(t, unknown, 6)
267+ require.Equal(t, 1, adoptable, "candidate must reach a terminal result exactly once")
150268 }
@@ -2,7 +2,9 @@ package snapshot
2 2
3 import (3 import (
4 "context"4 "context"
5+ "io"
5 "path/filepath"6 "path/filepath"
7+ "syscall"
6 "testing"8 "testing"
7 9
8 "github.com/bnema/vev/internal/domain"10 "github.com/bnema/vev/internal/domain"
@@ -12,6 +14,72 @@ import (
12 "github.com/stretchr/testify/require"14 "github.com/stretchr/testify/require"
13 )15 )
14 16
17+type fakeReconcileDrainDirectory struct {
18+ names []string
19+ position int
20+}
21+
22+func (d *fakeReconcileDrainDirectory) Seek(offset int64, whence int) (int64, error) {
23+ if whence == io.SeekCurrent {
24+ return int64(d.position), nil
25+ }
26+ d.position = int(offset)
27+ return offset, nil
28+}
29+
30+func (d *fakeReconcileDrainDirectory) ReadDirent(buffer []byte) (int, error) {
31+ if d.position >= len(d.names) {
32+ return 0, nil
33+ }
34+ batch := maintenanceDirentBatch(d.names[d.position:]...)
35+ d.position = len(d.names)
36+ return copy(buffer, batch), nil
37+}
38+
39+func (d *fakeReconcileDrainDirectory) Close() error { return nil }
40+
41+func configureDrainingReconcileScanner(repo *Repository, names []string) {
42+ repo.hooks.openMaintenanceDirectory = func(string) (maintenanceDirectory, error) {
43+ return &fakeReconcileDrainDirectory{names: names}, nil
44+ }
45+ repo.hooks.maintenanceDirentConfig = &maintenanceDirentConfig{
46+ drainBuffer: true,
47+ cookie: func(file maintenanceDirectory, _ *syscall.Dirent, remaining int) (int64, error) {
48+ end, err := file.Seek(0, io.SeekCurrent)
49+ return end - int64(remaining), err
50+ },
51+ }
52+}
53+
54+func TestReconciliationDrainedBatchHasNoSkippedOrDuplicateEntries(t *testing.T) {
55+ repo := NewRepository(privateDir(t))
56+ sessions := filepath.Join(repo.dir, repositorySessionsDir)
57+ names := make([]string, 3)
58+ for i := range names {
59+ id := domain.IncarnationID{byte(i + 20)}
60+ names[i] = id.String()
61+ require.NoError(t, safedir.EnsurePrivate(filepath.Join(sessions, names[i])))
62+ }
63+ configureDrainingReconcileScanner(repo, names)
64+
65+ cursor := ports.ReconcileCursor{}
66+ var got []string
67+ for calls := 0; ; calls++ {
68+ next, findings, err := repo.Reconcile(context.Background(), nil, cursor, ports.MaintenanceBudget{Entries: 2, Bytes: 1})
69+ require.NoError(t, err)
70+ for _, finding := range findings {
71+ got = append(got, finding.Candidate.Name)
72+ }
73+ if next.DirectoryCookie == 0 {
74+ break
75+ }
76+ require.Greater(t, next.DirectoryCookie, cursor.DirectoryCookie)
77+ cursor = next
78+ require.Less(t, calls, 3)
79+ }
80+ require.Equal(t, names, got)
81+}
82+
15 func TestReconciliationReadsEachCandidatePayloadOnceWithinBudget(t *testing.T) {83 func TestReconciliationReadsEachCandidatePayloadOnceWithinBudget(t *testing.T) {
16 repo := NewRepository(privateDir(t))84 repo := NewRepository(privateDir(t))
17 first := publishMaintenanceGenerations(t, repo, "first", 2)85 first := publishMaintenanceGenerations(t, repo, "first", 2)
@@ -94,6 +162,50 @@ func TestReconciliationReadsEachCandidatePayloadOnceWithinBudget(t *testing.T) {
94 }162 }
95 }163 }
96 164
165+func TestReconciliationIntrinsicallyOversizedCandidateMakesFixedBudgetProgress(t *testing.T) {
166+ tests := []struct {
167+ name string
168+ budget ports.MaintenanceBudget
169+ }{
170+ {name: "entry cost", budget: ports.MaintenanceBudget{Entries: 1, Bytes: 8 << 20}},
171+ {name: "byte cost", budget: ports.MaintenanceBudget{Entries: 8, Bytes: 1}},
172+ }
173+ for _, tc := range tests {
174+ t.Run(tc.name, func(t *testing.T) {
175+ repo := NewRepository(privateDir(t))
176+ publications := publishMaintenanceGenerations(t, repo, "work", 2)
177+ committed := checkpointRefForPublication(publications[0])
178+ record := domain.CatalogueRecord{Name: "work", IncarnationID: publications[0].IncarnationID, RecoveryState: domain.RecoveryHealthy, Committed: &committed}
179+ unknown := domain.IncarnationID{99}
180+ require.NoError(t, safedir.EnsurePrivate(filepath.Join(repo.dir, repositorySessionsDir, unknown.String())))
181+ configureDrainingReconcileScanner(repo, []string{record.IncarnationID.String(), unknown.String()})
182+
183+ next, findings, err := repo.Reconcile(context.Background(), []domain.CatalogueRecord{record}, ports.ReconcileCursor{}, tc.budget)
184+ require.NoError(t, err)
185+ require.Equal(t, uint64(1), next.DirectoryCookie)
186+ require.Len(t, findings, 1)
187+ require.Equal(t, ports.ReconcileBudgetExhausted, findings[0].Status)
188+ require.LessOrEqual(t, findings[0].Consumed.Entries, tc.budget.Entries)
189+ require.LessOrEqual(t, findings[0].Consumed.Bytes, tc.budget.Bytes)
190+
191+ later, findings, err := repo.Reconcile(context.Background(), []domain.CatalogueRecord{record}, next, tc.budget)
192+ require.NoError(t, err)
193+ require.Len(t, findings, 1)
194+ require.Equal(t, unknown, findings[0].Candidate.IncarnationID)
195+ require.Equal(t, ports.ReconcileQuarantined, findings[0].Status)
196+ if later.DirectoryCookie == 0 {
197+ return
198+ }
199+ require.Greater(t, later.DirectoryCookie, next.DirectoryCookie)
200+
201+ done, findings, err := repo.Reconcile(context.Background(), []domain.CatalogueRecord{record}, later, tc.budget)
202+ require.NoError(t, err)
203+ require.Zero(t, done)
204+ require.Empty(t, findings, "oversized candidate must not be retried")
205+ })
206+ }
207+}
208+
97 func TestReconciliationCursor(t *testing.T) {209 func TestReconciliationCursor(t *testing.T) {
98 repo := NewRepository(privateDir(t))210 repo := NewRepository(privateDir(t))
99 publications := publishMaintenanceGenerations(t, repo, "work", 2)211 publications := publishMaintenanceGenerations(t, repo, "work", 2)
@@ -132,19 +244,25 @@ func TestReconciliationCursor(t *testing.T) {
132 }244 }
133 require.Greater(t, calls, 1, "top-level traversal must resume across bounded calls")245 require.Greater(t, calls, 1, "top-level traversal must resume across bounded calls")
134 246
135- unknown := 0247+ unknown := make(map[domain.IncarnationID]struct{})
136 adoptable := 0248 adoptable := 0
137 for _, finding := range findings {249 for _, finding := range findings {
138 switch finding.Kind {250 switch finding.Kind {
139 case ports.ReconcileUnknownIncarnation:251 case ports.ReconcileUnknownIncarnation:
140- unknown++252+ require.Equal(t, ports.ReconcileQuarantined, finding.Status)
253+ _, duplicate := unknown[finding.Candidate.IncarnationID]
254+ require.False(t, duplicate, "unknown incarnation was returned twice")
255+ unknown[finding.Candidate.IncarnationID] = struct{}{}
141 case ports.ReconcileForwardOrphan:256 case ports.ReconcileForwardOrphan:
257+ if finding.Status == ports.ReconcileBudgetExhausted {
258+ continue
259+ }
142 require.Equal(t, ports.ReconcileValidated, finding.Status)260 require.Equal(t, ports.ReconcileValidated, finding.Status)
143 require.Equal(t, uint64(2), finding.Candidate.Ref.Generation)261 require.Equal(t, uint64(2), finding.Candidate.Ref.Generation)
144 require.Equal(t, committed, *finding.Candidate.Parent)262 require.Equal(t, committed, *finding.Candidate.Parent)
145 adoptable++263 adoptable++
146 }264 }
147 }265 }
148- require.Equal(t, 6, unknown)266+ require.Len(t, unknown, 6)
149- require.Equal(t, 1, adoptable)267+ require.Equal(t, 1, adoptable, "candidate must reach a terminal result exactly once")
150 }268 }
M internal/adapters/snapshot/repository.go +1 -0
@@ -85,6 +85,7 @@ type repositoryHooks struct {
8585 afterOpenRoot func()
8686 closeRoot func() error
8787 openMaintenanceDirectory func(string) (maintenanceDirectory, error)
88+ maintenanceDirentConfig *maintenanceDirentConfig
8889 beforeDirectoryRead func(string)
8990 beforeDeletionTombstoneRead func(string)
9091 // afterDeletionIncarnationQuarantine and afterDeletionLegacyQuarantine expose
@@ -85,6 +85,7 @@ type repositoryHooks struct {
85 afterOpenRoot func()85 afterOpenRoot func()
86 closeRoot func() error86 closeRoot func() error
87 openMaintenanceDirectory func(string) (maintenanceDirectory, error)87 openMaintenanceDirectory func(string) (maintenanceDirectory, error)
88+ maintenanceDirentConfig *maintenanceDirentConfig
88 beforeDirectoryRead func(string)89 beforeDirectoryRead func(string)
89 beforeDeletionTombstoneRead func(string)90 beforeDeletionTombstoneRead func(string)
90 // afterDeletionIncarnationQuarantine and afterDeletionLegacyQuarantine expose91 // afterDeletionIncarnationQuarantine and afterDeletionLegacyQuarantine expose
M internal/app/run_test.go +4 -1
@@ -466,7 +466,10 @@ func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) {
466466 })
467467
468468 t.Run("proven absence creates the catalogue before socket publication", func(t *testing.T) {
469- runtimeRoot, stateRoot := t.TempDir(), t.TempDir()
469+ runtimeRoot, err := os.MkdirTemp("/tmp", "vev-")
470+ require.NoError(t, err)
471+ t.Cleanup(func() { require.NoError(t, os.RemoveAll(runtimeRoot)) })
472+ stateRoot := t.TempDir()
470473 runtimeDir := filepath.Join(runtimeRoot, "vev")
471474 stateDir := filepath.Join(stateRoot, "vev")
472475 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
@@ -466,7 +466,10 @@ func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) {
466 })466 })
467 467
468 t.Run("proven absence creates the catalogue before socket publication", func(t *testing.T) {468 t.Run("proven absence creates the catalogue before socket publication", func(t *testing.T) {
469- runtimeRoot, stateRoot := t.TempDir(), t.TempDir()469+ runtimeRoot, err := os.MkdirTemp("/tmp", "vev-")
470+ require.NoError(t, err)
471+ t.Cleanup(func() { require.NoError(t, os.RemoveAll(runtimeRoot)) })
472+ stateRoot := t.TempDir()
470 runtimeDir := filepath.Join(runtimeRoot, "vev")473 runtimeDir := filepath.Join(runtimeRoot, "vev")
471 stateDir := filepath.Join(stateRoot, "vev")474 stateDir := filepath.Join(stateRoot, "vev")
472 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)475 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close