fix(media): remove only owned outputs, discard on abort and use context errors

commit: be238fd8af131543ce343f73c926525d98225bcc
author: brice · 22d ago · 2026-09-30 08:31 UTC
committer: brice · 22d ago · 2026-09-30 14:48 UTC
parent: 8d9898d
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

8 files · +416 -68

M .mockery.yml +2 -0


@@ -26,3 +26,5 @@ packages:
26	26		filename: io_mock_test.go
27	27		interfaces:
28	28		Writer:
	29	+	config:
	30	+	structname: MockIOWriter


@@ -26,3 +26,5 @@ packages:
26	filename: io_mock_test.go	26		filename: io_mock_test.go
27	interfaces:	27		interfaces:
28	Writer:	28		Writer:
		29	+	config:
		30	+	structname: MockIOWriter

A internal/adapters/capturefile/capturefile.go +60 -0

@@ -0,0 +1,60 @@
	1	+	// Package capturefile reserves capture output paths and removes only the file
	2	+	// that was reserved, never a different file that later took its place.
	3	+	package capturefile
	4	+
	5	+	import (
	6	+	"errors"
	7	+	"fmt"
	8	+	"io/fs"
	9	+	"os"
	10	+
	11	+	"github.com/bnema/nefercap/internal/ports"
	12	+	)
	13	+
	14	+	// Reservation identifies the file created by Create.
	15	+	type Reservation struct {
	16	+	path string
	17	+	info fs.FileInfo
	18	+	}
	19	+
	20	+	// Create makes a new private (0600) file at path with O_EXCL, so an existing
	21	+	// file or symlink is never opened or overwritten. An existing path yields an
	22	+	// error matching both ports.ErrPathExists and fs.ErrExist. The returned
	23	+	// Reservation records the file's identity for a later Remove.
	24	+	func Create(path string) (os.File, Reservation, error) {
	25	+	f, err := os.OpenFile(path, os.O_WRONLY\|os.O_CREATE\|os.O_EXCL, 0o600)
	26	+	if err != nil {
	27	+	if errors.Is(err, fs.ErrExist) {
	28	+	return nil, nil, fmt.Errorf("%w: %w", ports.ErrPathExists, err)
	29	+	}
	30	+	return nil, nil, fmt.Errorf("create output: %w", err)
	31	+	}
	32	+	info, err := f.Stat()
	33	+	if err != nil {
	34	+	_ = f.Close()
	35	+	_ = os.Remove(path) // just created by us; identity is unknown, so remove now
	36	+	return nil, nil, fmt.Errorf("stat output: %w", err)
	37	+	}
	38	+	return f, &Reservation{path: path, info: info}, nil
	39	+	}
	40	+
	41	+	// Remove deletes the reserved file if path still names it. A missing path is
	42	+	// success; a path that now names a different file (the original was moved and
	43	+	// replaced) is left untouched and is not an error. There is an unavoidable
	44	+	// short window between the identity check and the unlink.
	45	+	func (r *Reservation) Remove() error {
	46	+	cur, err := os.Lstat(r.path)
	47	+	if err != nil {
	48	+	if errors.Is(err, fs.ErrNotExist) {
	49	+	return nil
	50	+	}
	51	+	return fmt.Errorf("inspect output: %w", err)
	52	+	}
	53	+	if !os.SameFile(r.info, cur) {
	54	+	return nil
	55	+	}
	56	+	if err := os.Remove(r.path); err != nil && !errors.Is(err, fs.ErrNotExist) {
	57	+	return fmt.Errorf("remove output: %w", err)
	58	+	}
	59	+	return nil
	60	+	}

@@ -0,0 +1,60 @@
	1	+	// Package capturefile reserves capture output paths and removes only the file
	2	+	// that was reserved, never a different file that later took its place.
	3	+	package capturefile
	4	+
	5	+	import (
	6	+	"errors"
	7	+	"fmt"
	8	+	"io/fs"
	9	+	"os"
	10	+
	11	+	"github.com/bnema/nefercap/internal/ports"
	12	+	)
	13	+
	14	+	// Reservation identifies the file created by Create.
	15	+	type Reservation struct {
	16	+	path string
	17	+	info fs.FileInfo
	18	+	}
	19	+
	20	+	// Create makes a new private (0600) file at path with O_EXCL, so an existing
	21	+	// file or symlink is never opened or overwritten. An existing path yields an
	22	+	// error matching both ports.ErrPathExists and fs.ErrExist. The returned
	23	+	// Reservation records the file's identity for a later Remove.
	24	+	func Create(path string) (os.File, Reservation, error) {
	25	+	f, err := os.OpenFile(path, os.O_WRONLY\|os.O_CREATE\|os.O_EXCL, 0o600)
	26	+	if err != nil {
	27	+	if errors.Is(err, fs.ErrExist) {
	28	+	return nil, nil, fmt.Errorf("%w: %w", ports.ErrPathExists, err)
	29	+	}
	30	+	return nil, nil, fmt.Errorf("create output: %w", err)
	31	+	}
	32	+	info, err := f.Stat()
	33	+	if err != nil {
	34	+	_ = f.Close()
	35	+	_ = os.Remove(path) // just created by us; identity is unknown, so remove now
	36	+	return nil, nil, fmt.Errorf("stat output: %w", err)
	37	+	}
	38	+	return f, &Reservation{path: path, info: info}, nil
	39	+	}
	40	+
	41	+	// Remove deletes the reserved file if path still names it. A missing path is
	42	+	// success; a path that now names a different file (the original was moved and
	43	+	// replaced) is left untouched and is not an error. There is an unavoidable
	44	+	// short window between the identity check and the unlink.
	45	+	func (r *Reservation) Remove() error {
	46	+	cur, err := os.Lstat(r.path)
	47	+	if err != nil {
	48	+	if errors.Is(err, fs.ErrNotExist) {
	49	+	return nil
	50	+	}
	51	+	return fmt.Errorf("inspect output: %w", err)
	52	+	}
	53	+	if !os.SameFile(r.info, cur) {
	54	+	return nil
	55	+	}
	56	+	if err := os.Remove(r.path); err != nil && !errors.Is(err, fs.ErrNotExist) {
	57	+	return fmt.Errorf("remove output: %w", err)
	58	+	}
	59	+	return nil
	60	+	}

A internal/adapters/capturefile/capturefile_test.go +95 -0

@@ -0,0 +1,95 @@
	1	+	package capturefile
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"io/fs"
	6	+	"os"
	7	+	"path/filepath"
	8	+	"testing"
	9	+
	10	+	"github.com/bnema/nefercap/internal/ports"
	11	+	)
	12	+
	13	+	func TestCreateIsExclusiveAndPrivate(t *testing.T) {
	14	+	path := filepath.Join(t.TempDir(), "a")
	15	+	f, res, err := Create(path)
	16	+	if err != nil {
	17	+	t.Fatal(err)
	18	+	}
	19	+	defer f.Close()
	20	+	if info, err := os.Stat(path); err != nil \|\| info.Mode().Perm() != 0o600 {
	21	+	t.Fatalf("mode = %v, err = %v; want 0600", info.Mode(), err)
	22	+	}
	23	+	_, _, err = Create(path)
	24	+	if !errors.Is(err, ports.ErrPathExists) \|\| !errors.Is(err, fs.ErrExist) {
	25	+	t.Fatalf("got %v, want both ErrPathExists and fs.ErrExist", err)
	26	+	}
	27	+	var pe *fs.PathError
	28	+	if !errors.As(err, &pe) {
	29	+	t.Fatalf("path error detail lost: %v", err)
	30	+	}
	31	+	if err := res.Remove(); err != nil {
	32	+	t.Fatal(err)
	33	+	}
	34	+	if _, err := os.Lstat(path); !errors.Is(err, fs.ErrNotExist) {
	35	+	t.Fatalf("own file not removed: %v", err)
	36	+	}
	37	+	if err := res.Remove(); err != nil {
	38	+	t.Fatalf("second Remove: %v", err)
	39	+	}
	40	+	}
	41	+
	42	+	func TestCreateRefusesSymlink(t *testing.T) {
	43	+	dir := t.TempDir()
	44	+	link := filepath.Join(dir, "link")
	45	+	if err := os.Symlink(filepath.Join(dir, "target"), link); err != nil {
	46	+	t.Skip(err)
	47	+	}
	48	+	if _, _, err := Create(link); !errors.Is(err, ports.ErrPathExists) {
	49	+	t.Fatalf("got %v", err)
	50	+	}
	51	+	if _, err := os.Lstat(filepath.Join(dir, "target")); !errors.Is(err, fs.ErrNotExist) {
	52	+	t.Fatalf("symlink target created: %v", err)
	53	+	}
	54	+	}
	55	+
	56	+	func TestCreateOtherErrorsAreNotPathExists(t *testing.T) {
	57	+	_, _, err := Create(filepath.Join(t.TempDir(), "missing", "a"))
	58	+	if err == nil \|\| errors.Is(err, ports.ErrPathExists) {
	59	+	t.Fatalf("got %v", err)
	60	+	}
	61	+	}
	62	+
	63	+	func TestRemoveLeavesReplacement(t *testing.T) {
	64	+	dir := t.TempDir()
	65	+	path := filepath.Join(dir, "a")
	66	+	f, res, err := Create(path)
	67	+	if err != nil {
	68	+	t.Fatal(err)
	69	+	}
	70	+	defer f.Close()
	71	+	moved := filepath.Join(dir, "moved")
	72	+	if err := os.Rename(path, moved); err != nil {
	73	+	t.Fatal(err)
	74	+	}
	75	+	for name, replace := range map[string]func() error{
	76	+	"file": func() error { return os.WriteFile(path, []byte("user"), 0o600) },
	77	+	"symlink": func() error { return os.Symlink(moved, path) },
	78	+	} {
	79	+	if err := replace(); err != nil {
	80	+	t.Fatal(err)
	81	+	}
	82	+	if err := res.Remove(); err != nil {
	83	+	t.Fatalf("%s: %v", name, err)
	84	+	}
	85	+	if _, err := os.Lstat(path); err != nil {
	86	+	t.Fatalf("%s replacement was removed: %v", name, err)
	87	+	}
	88	+	if err := os.Remove(path); err != nil {
	89	+	t.Fatal(err)
	90	+	}
	91	+	}
	92	+	if _, err := os.Lstat(moved); err != nil {
	93	+	t.Fatalf("moved original touched: %v", err)
	94	+	}
	95	+	}

@@ -0,0 +1,95 @@
	1	+	package capturefile
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"io/fs"
	6	+	"os"
	7	+	"path/filepath"
	8	+	"testing"
	9	+
	10	+	"github.com/bnema/nefercap/internal/ports"
	11	+	)
	12	+
	13	+	func TestCreateIsExclusiveAndPrivate(t *testing.T) {
	14	+	path := filepath.Join(t.TempDir(), "a")
	15	+	f, res, err := Create(path)
	16	+	if err != nil {
	17	+	t.Fatal(err)
	18	+	}
	19	+	defer f.Close()
	20	+	if info, err := os.Stat(path); err != nil \|\| info.Mode().Perm() != 0o600 {
	21	+	t.Fatalf("mode = %v, err = %v; want 0600", info.Mode(), err)
	22	+	}
	23	+	_, _, err = Create(path)
	24	+	if !errors.Is(err, ports.ErrPathExists) \|\| !errors.Is(err, fs.ErrExist) {
	25	+	t.Fatalf("got %v, want both ErrPathExists and fs.ErrExist", err)
	26	+	}
	27	+	var pe *fs.PathError
	28	+	if !errors.As(err, &pe) {
	29	+	t.Fatalf("path error detail lost: %v", err)
	30	+	}
	31	+	if err := res.Remove(); err != nil {
	32	+	t.Fatal(err)
	33	+	}
	34	+	if _, err := os.Lstat(path); !errors.Is(err, fs.ErrNotExist) {
	35	+	t.Fatalf("own file not removed: %v", err)
	36	+	}
	37	+	if err := res.Remove(); err != nil {
	38	+	t.Fatalf("second Remove: %v", err)
	39	+	}
	40	+	}
	41	+
	42	+	func TestCreateRefusesSymlink(t *testing.T) {
	43	+	dir := t.TempDir()
	44	+	link := filepath.Join(dir, "link")
	45	+	if err := os.Symlink(filepath.Join(dir, "target"), link); err != nil {
	46	+	t.Skip(err)
	47	+	}
	48	+	if _, _, err := Create(link); !errors.Is(err, ports.ErrPathExists) {
	49	+	t.Fatalf("got %v", err)
	50	+	}
	51	+	if _, err := os.Lstat(filepath.Join(dir, "target")); !errors.Is(err, fs.ErrNotExist) {
	52	+	t.Fatalf("symlink target created: %v", err)
	53	+	}
	54	+	}
	55	+
	56	+	func TestCreateOtherErrorsAreNotPathExists(t *testing.T) {
	57	+	_, _, err := Create(filepath.Join(t.TempDir(), "missing", "a"))
	58	+	if err == nil \|\| errors.Is(err, ports.ErrPathExists) {
	59	+	t.Fatalf("got %v", err)
	60	+	}
	61	+	}
	62	+
	63	+	func TestRemoveLeavesReplacement(t *testing.T) {
	64	+	dir := t.TempDir()
	65	+	path := filepath.Join(dir, "a")
	66	+	f, res, err := Create(path)
	67	+	if err != nil {
	68	+	t.Fatal(err)
	69	+	}
	70	+	defer f.Close()
	71	+	moved := filepath.Join(dir, "moved")
	72	+	if err := os.Rename(path, moved); err != nil {
	73	+	t.Fatal(err)
	74	+	}
	75	+	for name, replace := range map[string]func() error{
	76	+	"file": func() error { return os.WriteFile(path, []byte("user"), 0o600) },
	77	+	"symlink": func() error { return os.Symlink(moved, path) },
	78	+	} {
	79	+	if err := replace(); err != nil {
	80	+	t.Fatal(err)
	81	+	}
	82	+	if err := res.Remove(); err != nil {
	83	+	t.Fatalf("%s: %v", name, err)
	84	+	}
	85	+	if _, err := os.Lstat(path); err != nil {
	86	+	t.Fatalf("%s replacement was removed: %v", name, err)
	87	+	}
	88	+	if err := os.Remove(path); err != nil {
	89	+	t.Fatal(err)
	90	+	}
	91	+	}
	92	+	if _, err := os.Lstat(moved); err != nil {
	93	+	t.Fatalf("moved original touched: %v", err)
	94	+	}
	95	+	}

M internal/adapters/ffmpeg/ffmpeg.go +46 -31


@@ -16,7 +16,6 @@ import (
16	16		"errors"
17	17		"fmt"
18	18		"io"
19		-	"io/fs"
20	19		"os"
21	20		"os/exec"
22	21		"strconv"

@@ -25,15 +24,15 @@ import (
25	24		"syscall"
26	25		"time"
27	26
	27	+	"github.com/bnema/nefercap/internal/adapters/capturefile"
28	28		"github.com/bnema/nefercap/internal/ports"
29	29		)
30	30
31	31		const (
32	32		defaultBinary = "ffmpeg"
33	33		// closeTimeout bounds how long Close waits for the encoder to drain after
34		-	// EOF before killing it; abortTimeout does the same for Abort.
	34	+	// EOF before killing it.
35	35		closeTimeout = 15 * time.Second
36		-	abortTimeout = 3 * time.Second
37	36		// killTimeout bounds the wait for the reaper after a kill.
38	37		killTimeout = 5 * time.Second
39	38		// exitProbe bounds how long a failed pipe write waits for the exit status.

@@ -70,7 +69,6 @@ type failure struct {
70	69		type Writer struct {
71	70		binary string
72	71		closeTimeout time.Duration
73		-	abortTimeout time.Duration
74	72
75	73		state atomic.Int32
76	74		mu sync.Mutex // serializes state transitions

@@ -78,7 +76,7 @@ type Writer struct {
78	76		// Set by Start before state becomes running; read-only afterwards.
79	77		cmd *exec.Cmd
80	78		stdin *os.File
81		-	path string
	79	+	res *capturefile.Reservation
82	80		width int
83	81		height int
84	82		format ports.PixelFormat

@@ -114,7 +112,6 @@ func NewWithBinary(binary string) *Writer {
114	112		return &Writer{
115	113		binary: binary,
116	114		closeTimeout: closeTimeout,
117		-	abortTimeout: abortTimeout,
118	115		abortCh: make(chan struct{}),
119	116		termDone: make(chan struct{}),
120	117		}

@@ -194,20 +191,17 @@ func (w *Writer) Start(ctx context.Context, first ports.Frame, path string, s po
194	191		if err != nil {
195	192		return fmt.Errorf("ffmpeg: %w", err)
196	193		}
197		-	out, err := os.OpenFile(path, os.O_WRONLY\|os.O_CREATE\|os.O_EXCL, 0o600)
	194	+	out, res, err := capturefile.Create(path)
198	195		if err != nil {
199		-	if errors.Is(err, fs.ErrExist) {
200		-	return fmt.Errorf("%w: %s", ports.ErrPathExists, path)
201		-	}
202		-	return fmt.Errorf("ffmpeg: create output: %w", err)
	196	+	return fmt.Errorf("ffmpeg: %w", err)
203	197		}
204		-	// From here we own path (created exclusively) and remove it on failure.
	198	+	// From here we own the created file and remove it (not a replacement) on failure.
205	199		fail := func(err error, closers ...*os.File) error {
206	200		for _, c := range closers {
207	201		_ = c.Close()
208	202		}
209		-	if rerr := os.Remove(path); rerr != nil && !errors.Is(rerr, fs.ErrNotExist) {
210		-	err = errors.Join(err, fmt.Errorf("ffmpeg: remove output: %w", rerr))
	203	+	if rerr := res.Remove(); rerr != nil {
	204	+	err = errors.Join(err, fmt.Errorf("ffmpeg: %w", rerr))
211	205		}
212	206		return err
213	207		}

@@ -233,7 +227,7 @@ func (w *Writer) Start(ctx context.Context, first ports.Frame, path string, s po
233	227		_ = pr.Close()
234	228		_ = out.Close()
235	229
236		-	w.cmd, w.stdin, w.path, w.stderr = cmd, pw, path, stderr
	230	+	w.cmd, w.stdin, w.res, w.stderr = cmd, pw, res, stderr
237	231		w.width, w.height, w.format = first.Width, first.Height, first.Format
238	232		w.waitDone = make(chan struct{})
239	233		go func() { // sole reaper: runs Wait exactly once, also after any Kill

@@ -241,7 +235,7 @@ func (w *Writer) Start(ctx context.Context, first ports.Frame, path string, s po
241	235		close(w.waitDone)
242	236		}()
243	237		if ctx.Done() != nil {
244		-	w.startStop = context.AfterFunc(ctx, func() { w.interrupt(context.Cause(ctx)) })
	238	+	w.startStop = context.AfterFunc(ctx, func() { w.interrupt(ctx.Err()) })
245	239		}
246	240		w.state.Store(stateRunning)
247	241		return nil

@@ -275,7 +269,7 @@ func (w *Writer) watch(ctx context.Context) (gen uint64, ok bool) {
275	269		w.writeDone = done
276	270		w.writeStop = context.AfterFunc(ctx, func() {
277	271		if w.active.Load() == g {
278		-	w.interrupt(context.Cause(ctx))
	272	+	w.interrupt(ctx.Err())
279	273		}
280	274		})
281	275		}

@@ -382,9 +376,12 @@ func writeFull(dst io.Writer, p []byte) error {
382	376		// removed otherwise. Repeated calls return the first result.
383	377		func (w *Writer) Close() error { return w.terminate(false) }
384	378
385		-	// Abort interrupts a pending Write and stops the encoder with a shorter grace
386		-	// period than Close; a Close already draining is escalated to a kill. It is
387		-	// safe concurrently with Write and idempotent.
	379	+	// Abort discards the recording: it interrupts a pending Write, kills the
	380	+	// encoder and removes the output file, even if frames were already written and
	381	+	// the encoder was healthy. A Close that is still draining is escalated to a
	382	+	// kill. Use Close to keep a video. It returns nil unless the encoder could not
	383	+	// be reaped or the file could not be removed, is safe concurrently with Write
	384	+	// and is idempotent.
388	385		func (w *Writer) Abort() error { return w.terminate(true) }
389	386
390	387		func (w *Writer) terminate(abort bool) error {

@@ -413,8 +410,8 @@ func (w *Writer) terminate(abort bool) error {
413	410		} else {
414	411		<-w.termDone
415	412		}
416		-	if abort && errors.Is(w.termErr, ErrNoFrames) {
417		-	return nil // nothing was recorded, so there is nothing to lose
	413	+	if abort && !first {
	414	+	return nil // the first terminal call owns the result
418	415		}
419	416		return w.termErr
420	417		}

@@ -431,19 +428,17 @@ func (w *Writer) shutdown(abort bool) error {
431	428
432	429		_ = w.stdin.Close() // EOF: the encoder finalizes the file
433	430
434		-	limit := w.closeTimeout
435		-	escalate := w.abortCh
436	431		if abort {
437		-	limit, escalate = w.abortTimeout, nil
	432	+	return w.discard()
438	433		}
439		-	timer := time.NewTimer(limit)
	434	+	timer := time.NewTimer(w.closeTimeout)
440	435		defer timer.Stop()
441	436		var err error
442	437		select {
443	438		case <-w.waitDone:
444	439		case <-timer.C:
445		-	err = fmt.Errorf("ffmpeg: encoder did not finish within %s; killed", limit)
446		-	case <-escalate:
	440	+	err = fmt.Errorf("ffmpeg: encoder did not finish within %s; killed", w.closeTimeout)
	441	+	case <-w.abortCh:
447	442		err = errors.New("ffmpeg: shutdown aborted; encoder killed")
448	443		}
449	444		if err != nil {

@@ -475,13 +470,33 @@ func (w *Writer) shutdown(abort bool) error {
475	470		err = ErrNoFrames
476	471		}
477	472		if err != nil {
478		-	if rerr := os.Remove(w.path); rerr != nil && !errors.Is(rerr, fs.ErrNotExist) {
479		-	err = errors.Join(err, fmt.Errorf("ffmpeg: remove output: %w", rerr))
480		-	}
	473	+	err = errors.Join(err, w.removeOutput())
481	474		}
482	475		return err
483	476		}
484	477
	478	+	// discard kills the encoder, waits for the reaper and removes the output.
	479	+	func (w *Writer) discard() error {
	480	+	_ = w.cmd.Process.Kill() // ErrProcessDone is harmless
	481	+	var err error
	482	+	kt := time.NewTimer(killTimeout)
	483	+	defer kt.Stop()
	484	+	select {
	485	+	case <-w.waitDone:
	486	+	case <-kt.C:
	487	+	err = errors.New("ffmpeg: encoder not reaped after kill")
	488	+	}
	489	+	return errors.Join(err, w.removeOutput())
	490	+	}
	491	+
	492	+	// removeOutput deletes the file Start reserved, but not a replacement.
	493	+	func (w *Writer) removeOutput() error {
	494	+	if err := w.res.Remove(); err != nil {
	495	+	return fmt.Errorf("ffmpeg: %w", err)
	496	+	}
	497	+	return nil
	498	+	}
	499	+
485	500		// boundedBuffer keeps the first limit bytes written and discards the rest, so
486	501		// a chatty encoder cannot grow memory.
487	502		type boundedBuffer struct {

 	"errors"
 	"fmt"
 	"io"
-	"io/fs"
 	"os"
 	"os/exec"
 	"strconv"
 	"syscall"
 	"time"
+	"github.com/bnema/nefercap/internal/adapters/capturefile"
 	"github.com/bnema/nefercap/internal/ports"
 )
 const (
 	defaultBinary = "ffmpeg"
 	// closeTimeout bounds how long Close waits for the encoder to drain after
-	// EOF before killing it; abortTimeout does the same for Abort.
+	// EOF before killing it.
 	closeTimeout = 15 * time.Second
-	abortTimeout = 3 * time.Second
 	// killTimeout bounds the wait for the reaper after a kill.
 	killTimeout = 5 * time.Second
 	// exitProbe bounds how long a failed pipe write waits for the exit status.
 type Writer struct {
 	binary       string
 	closeTimeout time.Duration
-	abortTimeout time.Duration
 	state atomic.Int32
 	mu    sync.Mutex // serializes state transitions
 	// Set by Start before state becomes running; read-only afterwards.
 	cmd      *exec.Cmd
 	stdin    *os.File
-	path     string
+	res      *capturefile.Reservation
 	width    int
 	height   int
 	format   ports.PixelFormat
 	return &Writer{
 		binary:       binary,
 		closeTimeout: closeTimeout,
-		abortTimeout: abortTimeout,
 		abortCh:      make(chan struct{}),
 		termDone:     make(chan struct{}),
 	}
 	if err != nil {
 		return fmt.Errorf("ffmpeg: %w", err)
 	}
-	out, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
+	out, res, err := capturefile.Create(path)
 	if err != nil {
-		if errors.Is(err, fs.ErrExist) {
+		return fmt.Errorf("ffmpeg: %w", err)
-			return fmt.Errorf("%w: %s", ports.ErrPathExists, path)
-		}
-		return fmt.Errorf("ffmpeg: create output: %w", err)
 	}
-	// From here we own path (created exclusively) and remove it on failure.
+	// From here we own the created file and remove it (not a replacement) on failure.
 	fail := func(err error, closers ...*os.File) error {
 		for _, c := range closers {
 			_ = c.Close()
 		}
-		if rerr := os.Remove(path); rerr != nil && !errors.Is(rerr, fs.ErrNotExist) {
+		if rerr := res.Remove(); rerr != nil {
-			err = errors.Join(err, fmt.Errorf("ffmpeg: remove output: %w", rerr))
+			err = errors.Join(err, fmt.Errorf("ffmpeg: %w", rerr))
 		}
 		return err
 	}
 	_ = pr.Close()
 	_ = out.Close()
-	w.cmd, w.stdin, w.path, w.stderr = cmd, pw, path, stderr
+	w.cmd, w.stdin, w.res, w.stderr = cmd, pw, res, stderr
 	w.width, w.height, w.format = first.Width, first.Height, first.Format
 	w.waitDone = make(chan struct{})
 	go func() { // sole reaper: runs Wait exactly once, also after any Kill
 		close(w.waitDone)
 	}()
 	if ctx.Done() != nil {
-		w.startStop = context.AfterFunc(ctx, func() { w.interrupt(context.Cause(ctx)) })
+		w.startStop = context.AfterFunc(ctx, func() { w.interrupt(ctx.Err()) })
 	}
 	w.state.Store(stateRunning)
 	return nil
 		w.writeDone = done
 		w.writeStop = context.AfterFunc(ctx, func() {
 			if w.active.Load() == g {
-				w.interrupt(context.Cause(ctx))
+				w.interrupt(ctx.Err())
 			}
 		})
 	}
 // removed otherwise. Repeated calls return the first result.
 func (w *Writer) Close() error { return w.terminate(false) }
-// Abort interrupts a pending Write and stops the encoder with a shorter grace
+// Abort discards the recording: it interrupts a pending Write, kills the
-// period than Close; a Close already draining is escalated to a kill. It is
+// encoder and removes the output file, even if frames were already written and
-// safe concurrently with Write and idempotent.
+// the encoder was healthy. A Close that is still draining is escalated to a
+// kill. Use Close to keep a video. It returns nil unless the encoder could not
+// be reaped or the file could not be removed, is safe concurrently with Write
+// and is idempotent.
 func (w *Writer) Abort() error { return w.terminate(true) }
 func (w *Writer) terminate(abort bool) error {
 	} else {
 		<-w.termDone
 	}
-	if abort && errors.Is(w.termErr, ErrNoFrames) {
+	if abort && !first {
-		return nil // nothing was recorded, so there is nothing to lose
+		return nil // the first terminal call owns the result
 	}
 	return w.termErr
 }
 	_ = w.stdin.Close() // EOF: the encoder finalizes the file
-	limit := w.closeTimeout
-	escalate := w.abortCh
 	if abort {
-		limit, escalate = w.abortTimeout, nil
+		return w.discard()
 	}
-	timer := time.NewTimer(limit)
+	timer := time.NewTimer(w.closeTimeout)
 	defer timer.Stop()
 	var err error
 	select {
 	case <-w.waitDone:
 	case <-timer.C:
-		err = fmt.Errorf("ffmpeg: encoder did not finish within %s; killed", limit)
+		err = fmt.Errorf("ffmpeg: encoder did not finish within %s; killed", w.closeTimeout)
-	case <-escalate:
+	case <-w.abortCh:
 		err = errors.New("ffmpeg: shutdown aborted; encoder killed")
 	}
 	if err != nil {
 		err = ErrNoFrames
 	}
 	if err != nil {
-		if rerr := os.Remove(w.path); rerr != nil && !errors.Is(rerr, fs.ErrNotExist) {
+		err = errors.Join(err, w.removeOutput())
-			err = errors.Join(err, fmt.Errorf("ffmpeg: remove output: %w", rerr))
-		}
 	}
 	return err
 }
+// discard kills the encoder, waits for the reaper and removes the output.
+func (w *Writer) discard() error {
+	_ = w.cmd.Process.Kill() // ErrProcessDone is harmless
+	var err error
+	kt := time.NewTimer(killTimeout)
+	defer kt.Stop()
+	select {
+	case <-w.waitDone:
+	case <-kt.C:
+		err = errors.New("ffmpeg: encoder not reaped after kill")
+	}
+	return errors.Join(err, w.removeOutput())
+}
+// removeOutput deletes the file Start reserved, but not a replacement.
+func (w *Writer) removeOutput() error {
+	if err := w.res.Remove(); err != nil {
+		return fmt.Errorf("ffmpeg: %w", err)
+	}
+	return nil
+}
 // boundedBuffer keeps the first limit bytes written and discards the rest, so
 // a chatty encoder cannot grow memory.
 type boundedBuffer struct {

M internal/adapters/ffmpeg/ffmpeg_test.go +155 -6


@@ -58,7 +58,6 @@ func helperWriter(t testing.T, mode string) Writer {
58	58		t.Setenv(helperEnv, mode)
59	59		w := NewWithBinary(os.Args[0])
60	60		w.closeTimeout = 300 * time.Millisecond
61		-	w.abortTimeout = 300 * time.Millisecond
62	61		t.Cleanup(func() { _ = w.Abort() })
63	62		return w
64	63		}

@@ -173,8 +172,8 @@ func TestStartRefusesExistingPath(t *testing.T) {
173	172		}
174	173		w := helperWriter(t, "drain")
175	174		err := w.Start(context.Background(), newFrame(4, 4, 16, 0), path, settings())
176		-	if !errors.Is(err, ports.ErrPathExists) {
177		-	t.Fatalf("got %v, want ErrPathExists", err)
	175	+	if !errors.Is(err, ports.ErrPathExists) \|\| !errors.Is(err, os.ErrExist) {
	176	+	t.Fatalf("got %v, want ErrPathExists and fs.ErrExist", err)
178	177		}
179	178		if got, _ := os.ReadFile(path); string(got) != "keep" {
180	179		t.Fatalf("existing file changed: %q", got)

@@ -482,7 +481,7 @@ func TestNeverStartedTerminalCallsAreNoOps(t *testing.T) {
482	481		}
483	482		}
484	483
485		-	func TestWriteAfterCloseDoesNotLeakWatchers(t *testing.T) {
	484	+	func TestWatcherRegisteredOncePerContext(t *testing.T) {
486	485		w := helperWriter(t, "drain")
487	486		w.closeTimeout = 10 * time.Second
488	487		f := newFrame(4, 4, 16, 0)

@@ -503,6 +502,156 @@ func TestWriteAfterCloseDoesNotLeakWatchers(t *testing.T) {
503	502		t.Fatal(err)
504	503		}
505	504		cancel() // after Close: must be harmless
	505	+	if err := w.Write(ctx, f); !errors.Is(err, ports.ErrClosed) {
	506	+	t.Fatalf("Write after Close: %v, want ErrClosed", err)
	507	+	}
	508	+	if w.writeGen != 1 {
	509	+	t.Fatal("Write after Close registered a watcher")
	510	+	}
	511	+	}
	512	+
	513	+	func TestWriteContextAlternation(t *testing.T) {
	514	+	w := helperWriter(t, "hang")
	515	+	f := newFrame(64, 64, 256, 0) // 16 KiB: one frame fits the pipe, many do not
	516	+	session, cancelSession := context.WithCancel(context.Background())
	517	+	defer cancelSession()
	518	+	if err := w.Start(session, f, filepath.Join(t.TempDir(), "v.mp4"), settings()); err != nil {
	519	+	t.Fatal(err)
	520	+	}
	521	+	ctxA, cancelA := context.WithCancel(context.Background())
	522	+	ctxB, cancelB := context.WithCancel(context.Background())
	523	+	defer cancelB()
	524	+	if err := w.Write(ctxA, f); err != nil {
	525	+	t.Fatal(err)
	526	+	}
	527	+	res := make(chan error, 1)
	528	+	go func() { // fill the pipe until a Write blocks on the hung encoder
	529	+	for {
	530	+	if err := w.Write(ctxB, f); err != nil {
	531	+	res <- err
	532	+	return
	533	+	}
	534	+	}
	535	+	}()
	536	+	time.Sleep(300 * time.Millisecond)
	537	+	cancelA() // an earlier context must not interrupt B's blocked Write
	538	+	select {
	539	+	case err := <-res:
	540	+	t.Fatalf("cancelling A interrupted B: %v", err)
	541	+	case <-time.After(300 * time.Millisecond):
	542	+	}
	543	+	if w.cause.Load() != nil {
	544	+	t.Fatal("cancelling A poisoned the writer")
	545	+	}
	546	+	cancelSession() // the Start context is the recording session and does interrupt
	547	+	select {
	548	+	case err := <-res:
	549	+	if !errors.Is(err, context.Canceled) {
	550	+	t.Fatalf("got %v, want Canceled", err)
	551	+	}
	552	+	case <-time.After(5 * time.Second):
	553	+	t.Fatal("session cancel did not interrupt B")
	554	+	}
	555	+	}
	556	+
	557	+	func TestCancelCauseNormalizedToContextErr(t *testing.T) {
	558	+	custom := errors.New("custom cause")
	559	+	f := newFrame(1024, 1024, 4096, 0)
	560	+	for _, where := range []string{"start", "write"} {
	561	+	t.Run(where, func(t *testing.T) {
	562	+	w := helperWriter(t, "hang")
	563	+	cause, cancel := context.WithCancelCause(context.Background())
	564	+	startCtx, writeCtx := context.Background(), context.Background()
	565	+	if where == "start" {
	566	+	startCtx = cause
	567	+	} else {
	568	+	writeCtx = cause
	569	+	}
	570	+	if err := w.Start(startCtx, f, filepath.Join(t.TempDir(), "v.mp4"), settings()); err != nil {
	571	+	t.Fatal(err)
	572	+	}
	573	+	res := make(chan error, 1)
	574	+	go func() { res <- w.Write(writeCtx, f) }()
	575	+	time.Sleep(200 * time.Millisecond)
	576	+	cancel(custom)
	577	+	select {
	578	+	case err := <-res:
	579	+	if !errors.Is(err, context.Canceled) \|\| errors.Is(err, custom) {
	580	+	t.Fatalf("got %v, want context.Canceled without the custom cause", err)
	581	+	}
	582	+	case <-time.After(5 * time.Second):
	583	+	t.Fatal("cancel did not interrupt Write")
	584	+	}
	585	+	// The sticky retry keeps the frozen-contract error.
	586	+	if err := w.Write(context.Background(), f); !errors.Is(err, context.Canceled) \|\| errors.Is(err, custom) {
	587	+	t.Fatalf("retry: got %v", err)
	588	+	}
	589	+	})
	590	+	}
	591	+	}
	592	+
	593	+	func TestAbortDiscardsHealthyRecording(t *testing.T) {
	594	+	w := helperWriter(t, "drain")
	595	+	path := filepath.Join(t.TempDir(), "v.mp4")
	596	+	f := newFrame(4, 4, 16, 0)
	597	+	if err := w.Start(context.Background(), f, path, settings()); err != nil {
	598	+	t.Fatal(err)
	599	+	}
	600	+	for i := 0; i < 3; i++ {
	601	+	if err := w.Write(context.Background(), f); err != nil {
	602	+	t.Fatal(err)
	603	+	}
	604	+	}
	605	+	pid := w.cmd.Process.Pid
	606	+	if err := w.Abort(); err != nil {
	607	+	t.Fatalf("Abort = %v, want nil", err)
	608	+	}
	609	+	if !gone(pid) {
	610	+	t.Fatal("encoder not reaped")
	611	+	}
	612	+	if _, err := os.Lstat(path); !errors.Is(err, os.ErrNotExist) {
	613	+	t.Fatalf("Abort kept the recording: %v", err)
	614	+	}
	615	+	if err := w.Close(); err != nil {
	616	+	t.Fatalf("Close after Abort = %v", err)
	617	+	}
	618	+	}
	619	+
	620	+	func TestTerminalCleanupSparesReplacementFile(t *testing.T) {
	621	+	for _, tc := range []struct {
	622	+	name string
	623	+	end func(*Writer) error
	624	+	}{
	625	+	{"Abort", func(w *Writer) error { return w.Abort() }},
	626	+	{"Close killing a stuck encoder", func(w *Writer) error { _ = w.Close(); return nil }},
	627	+	} {
	628	+	t.Run(tc.name, func(t *testing.T) {
	629	+	w := helperWriter(t, "slowclose")
	630	+	dir := t.TempDir()
	631	+	path := filepath.Join(dir, "v.mp4")
	632	+	f := newFrame(4, 4, 16, 0)
	633	+	if err := w.Start(context.Background(), f, path, settings()); err != nil {
	634	+	t.Fatal(err)
	635	+	}
	636	+	pid := w.cmd.Process.Pid
	637	+	moved := filepath.Join(dir, "moved.mp4")
	638	+	if err := os.Rename(path, moved); err != nil {
	639	+	t.Fatal(err)
	640	+	}
	641	+	if err := os.WriteFile(path, []byte("user data"), 0o600); err != nil {
	642	+	t.Fatal(err)
	643	+	}
	644	+	if err := tc.end(w); err != nil {
	645	+	t.Fatal(err)
	646	+	}
	647	+	if !gone(pid) {
	648	+	t.Fatal("encoder not reaped")
	649	+	}
	650	+	if got, err := os.ReadFile(path); err != nil \|\| string(got) != "user data" {
	651	+	t.Fatalf("replacement was removed or changed: %q, %v", got, err)
	652	+	}
	653	+	})
	654	+	}
506	655		}
507	656
508	657		func TestRealFFmpeg(t *testing.T) {

@@ -705,7 +854,7 @@ func TestWriteRowsShortWrite(t *testing.T) {
705	854		calls int
706	855		}{"packed": {packed, 1}, "rows": {padded, 1}} {
707	856		t.Run(name, func(t *testing.T) {
708		-	m := NewMockWriter(t)
	857	+	m := NewMockIOWriter(t)
709	858		m.EXPECT().Write(mock.Anything).Return(3, nil).Times(tc.calls)
710	859		if err := writeRows(m, tc.f); !errors.Is(err, io.ErrShortWrite) {
711	860		t.Fatalf("got %v, want io.ErrShortWrite", err)

@@ -714,7 +863,7 @@ func TestWriteRowsShortWrite(t *testing.T) {
714	863		}
715	864		t.Run("error passes through", func(t *testing.T) {
716	865		boom := errors.New("boom")
717		-	m := NewMockWriter(t)
	866	+	m := NewMockIOWriter(t)
718	867		m.EXPECT().Write(mock.Anything).Return(0, boom).Once()
719	868		if err := writeRows(m, padded); !errors.Is(err, boom) {
720	869		t.Fatalf("got %v", err)


@@ -58,7 +58,6 @@ func helperWriter(t testing.T, mode string) Writer {
58		t.Setenv(helperEnv, mode)	58		t.Setenv(helperEnv, mode)
59		w := NewWithBinary(os.Args[0])	59		w := NewWithBinary(os.Args[0])
60		w.closeTimeout = 300 * time.Millisecond	60		w.closeTimeout = 300 * time.Millisecond
61	-	w.abortTimeout = 300 * time.Millisecond
62		t.Cleanup(func() { _ = w.Abort() })	61		t.Cleanup(func() { _ = w.Abort() })
63		return w	62		return w
64		}	63		}

@@ -173,8 +172,8 @@ func TestStartRefusesExistingPath(t *testing.T) {
173		}	172		}
174		w := helperWriter(t, "drain")	173		w := helperWriter(t, "drain")
175		err := w.Start(context.Background(), newFrame(4, 4, 16, 0), path, settings())	174		err := w.Start(context.Background(), newFrame(4, 4, 16, 0), path, settings())
176	-	if !errors.Is(err, ports.ErrPathExists) {	175	+	if !errors.Is(err, ports.ErrPathExists) \|\| !errors.Is(err, os.ErrExist) {
177	-	t.Fatalf("got %v, want ErrPathExists", err)	176	+	t.Fatalf("got %v, want ErrPathExists and fs.ErrExist", err)
178		}	177		}
179		if got, _ := os.ReadFile(path); string(got) != "keep" {	178		if got, _ := os.ReadFile(path); string(got) != "keep" {
180		t.Fatalf("existing file changed: %q", got)	179		t.Fatalf("existing file changed: %q", got)

@@ -482,7 +481,7 @@ func TestNeverStartedTerminalCallsAreNoOps(t *testing.T) {
482		}	481		}
483		}	482		}
484			483
485	-	func TestWriteAfterCloseDoesNotLeakWatchers(t *testing.T) {	484	+	func TestWatcherRegisteredOncePerContext(t *testing.T) {
486		w := helperWriter(t, "drain")	485		w := helperWriter(t, "drain")
487		w.closeTimeout = 10 * time.Second	486		w.closeTimeout = 10 * time.Second
488		f := newFrame(4, 4, 16, 0)	487		f := newFrame(4, 4, 16, 0)

@@ -503,6 +502,156 @@ func TestWriteAfterCloseDoesNotLeakWatchers(t *testing.T) {
503		t.Fatal(err)	502		t.Fatal(err)
504		}	503		}
505		cancel() // after Close: must be harmless	504		cancel() // after Close: must be harmless
			505	+	if err := w.Write(ctx, f); !errors.Is(err, ports.ErrClosed) {
			506	+	t.Fatalf("Write after Close: %v, want ErrClosed", err)
			507	+	}
			508	+	if w.writeGen != 1 {
			509	+	t.Fatal("Write after Close registered a watcher")
			510	+	}
			511	+	}
			512	+
			513	+	func TestWriteContextAlternation(t *testing.T) {
			514	+	w := helperWriter(t, "hang")
			515	+	f := newFrame(64, 64, 256, 0) // 16 KiB: one frame fits the pipe, many do not
			516	+	session, cancelSession := context.WithCancel(context.Background())
			517	+	defer cancelSession()
			518	+	if err := w.Start(session, f, filepath.Join(t.TempDir(), "v.mp4"), settings()); err != nil {
			519	+	t.Fatal(err)
			520	+	}
			521	+	ctxA, cancelA := context.WithCancel(context.Background())
			522	+	ctxB, cancelB := context.WithCancel(context.Background())
			523	+	defer cancelB()
			524	+	if err := w.Write(ctxA, f); err != nil {
			525	+	t.Fatal(err)
			526	+	}
			527	+	res := make(chan error, 1)
			528	+	go func() { // fill the pipe until a Write blocks on the hung encoder
			529	+	for {
			530	+	if err := w.Write(ctxB, f); err != nil {
			531	+	res <- err
			532	+	return
			533	+	}
			534	+	}
			535	+	}()
			536	+	time.Sleep(300 * time.Millisecond)
			537	+	cancelA() // an earlier context must not interrupt B's blocked Write
			538	+	select {
			539	+	case err := <-res:
			540	+	t.Fatalf("cancelling A interrupted B: %v", err)
			541	+	case <-time.After(300 * time.Millisecond):
			542	+	}
			543	+	if w.cause.Load() != nil {
			544	+	t.Fatal("cancelling A poisoned the writer")
			545	+	}
			546	+	cancelSession() // the Start context is the recording session and does interrupt
			547	+	select {
			548	+	case err := <-res:
			549	+	if !errors.Is(err, context.Canceled) {
			550	+	t.Fatalf("got %v, want Canceled", err)
			551	+	}
			552	+	case <-time.After(5 * time.Second):
			553	+	t.Fatal("session cancel did not interrupt B")
			554	+	}
			555	+	}
			556	+
			557	+	func TestCancelCauseNormalizedToContextErr(t *testing.T) {
			558	+	custom := errors.New("custom cause")
			559	+	f := newFrame(1024, 1024, 4096, 0)
			560	+	for _, where := range []string{"start", "write"} {
			561	+	t.Run(where, func(t *testing.T) {
			562	+	w := helperWriter(t, "hang")
			563	+	cause, cancel := context.WithCancelCause(context.Background())
			564	+	startCtx, writeCtx := context.Background(), context.Background()
			565	+	if where == "start" {
			566	+	startCtx = cause
			567	+	} else {
			568	+	writeCtx = cause
			569	+	}
			570	+	if err := w.Start(startCtx, f, filepath.Join(t.TempDir(), "v.mp4"), settings()); err != nil {
			571	+	t.Fatal(err)
			572	+	}
			573	+	res := make(chan error, 1)
			574	+	go func() { res <- w.Write(writeCtx, f) }()
			575	+	time.Sleep(200 * time.Millisecond)
			576	+	cancel(custom)
			577	+	select {
			578	+	case err := <-res:
			579	+	if !errors.Is(err, context.Canceled) \|\| errors.Is(err, custom) {
			580	+	t.Fatalf("got %v, want context.Canceled without the custom cause", err)
			581	+	}
			582	+	case <-time.After(5 * time.Second):
			583	+	t.Fatal("cancel did not interrupt Write")
			584	+	}
			585	+	// The sticky retry keeps the frozen-contract error.
			586	+	if err := w.Write(context.Background(), f); !errors.Is(err, context.Canceled) \|\| errors.Is(err, custom) {
			587	+	t.Fatalf("retry: got %v", err)
			588	+	}
			589	+	})
			590	+	}
			591	+	}
			592	+
			593	+	func TestAbortDiscardsHealthyRecording(t *testing.T) {
			594	+	w := helperWriter(t, "drain")
			595	+	path := filepath.Join(t.TempDir(), "v.mp4")
			596	+	f := newFrame(4, 4, 16, 0)
			597	+	if err := w.Start(context.Background(), f, path, settings()); err != nil {
			598	+	t.Fatal(err)
			599	+	}
			600	+	for i := 0; i < 3; i++ {
			601	+	if err := w.Write(context.Background(), f); err != nil {
			602	+	t.Fatal(err)
			603	+	}
			604	+	}
			605	+	pid := w.cmd.Process.Pid
			606	+	if err := w.Abort(); err != nil {
			607	+	t.Fatalf("Abort = %v, want nil", err)
			608	+	}
			609	+	if !gone(pid) {
			610	+	t.Fatal("encoder not reaped")
			611	+	}
			612	+	if _, err := os.Lstat(path); !errors.Is(err, os.ErrNotExist) {
			613	+	t.Fatalf("Abort kept the recording: %v", err)
			614	+	}
			615	+	if err := w.Close(); err != nil {
			616	+	t.Fatalf("Close after Abort = %v", err)
			617	+	}
			618	+	}
			619	+
			620	+	func TestTerminalCleanupSparesReplacementFile(t *testing.T) {
			621	+	for _, tc := range []struct {
			622	+	name string
			623	+	end func(*Writer) error
			624	+	}{
			625	+	{"Abort", func(w *Writer) error { return w.Abort() }},
			626	+	{"Close killing a stuck encoder", func(w *Writer) error { _ = w.Close(); return nil }},
			627	+	} {
			628	+	t.Run(tc.name, func(t *testing.T) {
			629	+	w := helperWriter(t, "slowclose")
			630	+	dir := t.TempDir()
			631	+	path := filepath.Join(dir, "v.mp4")
			632	+	f := newFrame(4, 4, 16, 0)
			633	+	if err := w.Start(context.Background(), f, path, settings()); err != nil {
			634	+	t.Fatal(err)
			635	+	}
			636	+	pid := w.cmd.Process.Pid
			637	+	moved := filepath.Join(dir, "moved.mp4")
			638	+	if err := os.Rename(path, moved); err != nil {
			639	+	t.Fatal(err)
			640	+	}
			641	+	if err := os.WriteFile(path, []byte("user data"), 0o600); err != nil {
			642	+	t.Fatal(err)
			643	+	}
			644	+	if err := tc.end(w); err != nil {
			645	+	t.Fatal(err)
			646	+	}
			647	+	if !gone(pid) {
			648	+	t.Fatal("encoder not reaped")
			649	+	}
			650	+	if got, err := os.ReadFile(path); err != nil \|\| string(got) != "user data" {
			651	+	t.Fatalf("replacement was removed or changed: %q, %v", got, err)
			652	+	}
			653	+	})
			654	+	}
506		}	655		}
507			656
508		func TestRealFFmpeg(t *testing.T) {	657		func TestRealFFmpeg(t *testing.T) {

@@ -705,7 +854,7 @@ func TestWriteRowsShortWrite(t *testing.T) {
705		calls int	854		calls int
706		}{"packed": {packed, 1}, "rows": {padded, 1}} {	855		}{"packed": {packed, 1}, "rows": {padded, 1}} {
707		t.Run(name, func(t *testing.T) {	856		t.Run(name, func(t *testing.T) {
708	-	m := NewMockWriter(t)	857	+	m := NewMockIOWriter(t)
709		m.EXPECT().Write(mock.Anything).Return(3, nil).Times(tc.calls)	858		m.EXPECT().Write(mock.Anything).Return(3, nil).Times(tc.calls)
710		if err := writeRows(m, tc.f); !errors.Is(err, io.ErrShortWrite) {	859		if err := writeRows(m, tc.f); !errors.Is(err, io.ErrShortWrite) {
711		t.Fatalf("got %v, want io.ErrShortWrite", err)	860		t.Fatalf("got %v, want io.ErrShortWrite", err)

@@ -714,7 +863,7 @@ func TestWriteRowsShortWrite(t *testing.T) {
714		}	863		}
715		t.Run("error passes through", func(t *testing.T) {	864		t.Run("error passes through", func(t *testing.T) {
716		boom := errors.New("boom")	865		boom := errors.New("boom")
717	-	m := NewMockWriter(t)	866	+	m := NewMockIOWriter(t)
718		m.EXPECT().Write(mock.Anything).Return(0, boom).Once()	867		m.EXPECT().Write(mock.Anything).Return(0, boom).Once()
719		if err := writeRows(m, padded); !errors.Is(err, boom) {	868		if err := writeRows(m, padded); !errors.Is(err, boom) {
720		t.Fatalf("got %v", err)	869		t.Fatalf("got %v", err)

M internal/adapters/ffmpeg/io_mock_test.go +18 -18


@@ -8,17 +8,17 @@ import (
8	8		mock "github.com/stretchr/testify/mock"
9	9		)
10	10
11		-	// NewMockWriter creates a new instance of MockWriter. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
	11	+	// NewMockIOWriter creates a new instance of MockIOWriter. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
12	12		// The first argument is typically a *testing.T value.
13		-	func NewMockWriter(t interface {
	13	+	func NewMockIOWriter(t interface {
14	14		mock.TestingT
15	15		Cleanup(func())
16		-	}) *MockWriter {
	16	+	}) *MockIOWriter {
17	17		if helper, ok := t.(interface{ Helper() }); ok {
18	18		helper.Helper()
19	19		}
20	20
21		-	mock := &MockWriter{}
	21	+	mock := &MockIOWriter{}
22	22		mock.Mock.Test(t)
23	23
24	24		t.Cleanup(func() {

@@ -31,21 +31,21 @@ func NewMockWriter(t interface {
31	31		return mock
32	32		}
33	33
34		-	// MockWriter is an autogenerated mock type for the Writer type
35		-	type MockWriter struct {
	34	+	// MockIOWriter is an autogenerated mock type for the Writer type
	35	+	type MockIOWriter struct {
36	36		mock.Mock
37	37		}
38	38
39		-	type MockWriter_Expecter struct {
	39	+	type MockIOWriter_Expecter struct {
40	40		mock *mock.Mock
41	41		}
42	42
43		-	func (_m MockWriter) EXPECT() MockWriter_Expecter {
44		-	return &MockWriter_Expecter{mock: &_m.Mock}
	43	+	func (_m MockIOWriter) EXPECT() MockIOWriter_Expecter {
	44	+	return &MockIOWriter_Expecter{mock: &_m.Mock}
45	45		}
46	46
47		-	// Write provides a mock function for the type MockWriter
48		-	func (_mock *MockWriter) Write(p []byte) (int, error) {
	47	+	// Write provides a mock function for the type MockIOWriter
	48	+	func (_mock *MockIOWriter) Write(p []byte) (int, error) {
49	49		ret := _mock.Called(p)
50	50
51	51		if len(ret) == 0 {

@@ -70,18 +70,18 @@ func (_mock *MockWriter) Write(p []byte) (int, error) {
70	70		return r0, r1
71	71		}
72	72
73		-	// MockWriter_Write_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Write'
74		-	type MockWriter_Write_Call struct {
	73	+	// MockIOWriter_Write_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Write'
	74	+	type MockIOWriter_Write_Call struct {
75	75		*mock.Call
76	76		}
77	77
78	78		// Write is a helper method to define mock.On call
79	79		// - p []byte
80		-	func (_e MockWriter_Expecter) Write(p any) MockWriter_Write_Call {
81		-	return &MockWriter_Write_Call{Call: _e.mock.On("Write", p)}
	80	+	func (_e MockIOWriter_Expecter) Write(p any) MockIOWriter_Write_Call {
	81	+	return &MockIOWriter_Write_Call{Call: _e.mock.On("Write", p)}
82	82		}
83	83
84		-	func (_c MockWriter_Write_Call) Run(run func(p []byte)) MockWriter_Write_Call {
	84	+	func (_c MockIOWriter_Write_Call) Run(run func(p []byte)) MockIOWriter_Write_Call {
85	85		_c.Call.Run(func(args mock.Arguments) {
86	86		var arg0 []byte
87	87		if args[0] != nil {

@@ -94,12 +94,12 @@ func (_c MockWriter_Write_Call) Run(run func(p []byte)) MockWriter_Write_Call
94	94		return _c
95	95		}
96	96
97		-	func (_c MockWriter_Write_Call) Return(n int, err error) MockWriter_Write_Call {
	97	+	func (_c MockIOWriter_Write_Call) Return(n int, err error) MockIOWriter_Write_Call {
98	98		_c.Call.Return(n, err)
99	99		return _c
100	100		}
101	101
102		-	func (_c MockWriter_Write_Call) RunAndReturn(run func(p []byte) (int, error)) MockWriter_Write_Call {
	102	+	func (_c MockIOWriter_Write_Call) RunAndReturn(run func(p []byte) (int, error)) MockIOWriter_Write_Call {
103	103		_c.Call.Return(run)
104	104		return _c
105	105		}


@@ -8,17 +8,17 @@ import (
8		mock "github.com/stretchr/testify/mock"	8		mock "github.com/stretchr/testify/mock"
9		)	9		)
10			10
11	-	// NewMockWriter creates a new instance of MockWriter. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.	11	+	// NewMockIOWriter creates a new instance of MockIOWriter. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
12		// The first argument is typically a *testing.T value.	12		// The first argument is typically a *testing.T value.
13	-	func NewMockWriter(t interface {	13	+	func NewMockIOWriter(t interface {
14		mock.TestingT	14		mock.TestingT
15		Cleanup(func())	15		Cleanup(func())
16	-	}) *MockWriter {	16	+	}) *MockIOWriter {
17		if helper, ok := t.(interface{ Helper() }); ok {	17		if helper, ok := t.(interface{ Helper() }); ok {
18		helper.Helper()	18		helper.Helper()
19		}	19		}
20			20
21	-	mock := &MockWriter{}	21	+	mock := &MockIOWriter{}
22		mock.Mock.Test(t)	22		mock.Mock.Test(t)
23			23
24		t.Cleanup(func() {	24		t.Cleanup(func() {

@@ -31,21 +31,21 @@ func NewMockWriter(t interface {
31		return mock	31		return mock
32		}	32		}
33			33
34	-	// MockWriter is an autogenerated mock type for the Writer type	34	+	// MockIOWriter is an autogenerated mock type for the Writer type
35	-	type MockWriter struct {	35	+	type MockIOWriter struct {
36		mock.Mock	36		mock.Mock
37		}	37		}
38			38
39	-	type MockWriter_Expecter struct {	39	+	type MockIOWriter_Expecter struct {
40		mock *mock.Mock	40		mock *mock.Mock
41		}	41		}
42			42
43	-	func (_m MockWriter) EXPECT() MockWriter_Expecter {	43	+	func (_m MockIOWriter) EXPECT() MockIOWriter_Expecter {
44	-	return &MockWriter_Expecter{mock: &_m.Mock}	44	+	return &MockIOWriter_Expecter{mock: &_m.Mock}
45		}	45		}
46			46
47	-	// Write provides a mock function for the type MockWriter	47	+	// Write provides a mock function for the type MockIOWriter
48	-	func (_mock *MockWriter) Write(p []byte) (int, error) {	48	+	func (_mock *MockIOWriter) Write(p []byte) (int, error) {
49		ret := _mock.Called(p)	49		ret := _mock.Called(p)
50			50
51		if len(ret) == 0 {	51		if len(ret) == 0 {

@@ -70,18 +70,18 @@ func (_mock *MockWriter) Write(p []byte) (int, error) {
70		return r0, r1	70		return r0, r1
71		}	71		}
72			72
73	-	// MockWriter_Write_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Write'	73	+	// MockIOWriter_Write_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Write'
74	-	type MockWriter_Write_Call struct {	74	+	type MockIOWriter_Write_Call struct {
75		*mock.Call	75		*mock.Call
76		}	76		}
77			77
78		// Write is a helper method to define mock.On call	78		// Write is a helper method to define mock.On call
79		// - p []byte	79		// - p []byte
80	-	func (_e MockWriter_Expecter) Write(p any) MockWriter_Write_Call {	80	+	func (_e MockIOWriter_Expecter) Write(p any) MockIOWriter_Write_Call {
81	-	return &MockWriter_Write_Call{Call: _e.mock.On("Write", p)}	81	+	return &MockIOWriter_Write_Call{Call: _e.mock.On("Write", p)}
82		}	82		}
83			83
84	-	func (_c MockWriter_Write_Call) Run(run func(p []byte)) MockWriter_Write_Call {	84	+	func (_c MockIOWriter_Write_Call) Run(run func(p []byte)) MockIOWriter_Write_Call {
85		_c.Call.Run(func(args mock.Arguments) {	85		_c.Call.Run(func(args mock.Arguments) {
86		var arg0 []byte	86		var arg0 []byte
87		if args[0] != nil {	87		if args[0] != nil {

@@ -94,12 +94,12 @@ func (_c MockWriter_Write_Call) Run(run func(p []byte)) MockWriter_Write_Call
94		return _c	94		return _c
95		}	95		}
96			96
97	-	func (_c MockWriter_Write_Call) Return(n int, err error) MockWriter_Write_Call {	97	+	func (_c MockIOWriter_Write_Call) Return(n int, err error) MockIOWriter_Write_Call {
98		_c.Call.Return(n, err)	98		_c.Call.Return(n, err)
99		return _c	99		return _c
100		}	100		}
101			101
102	-	func (_c MockWriter_Write_Call) RunAndReturn(run func(p []byte) (int, error)) MockWriter_Write_Call {	102	+	func (_c MockIOWriter_Write_Call) RunAndReturn(run func(p []byte) (int, error)) MockIOWriter_Write_Call {
103		_c.Call.Return(run)	103		_c.Call.Return(run)
104		return _c	104		return _c
105		}	105		}

M internal/adapters/png/png.go +6 -11


@@ -9,9 +9,8 @@ import (
9	9		"image"
10	10		stdpng "image/png"
11	11		"io"
12		-	"io/fs"
13		-	"os"
14	12
	13	+	"github.com/bnema/nefercap/internal/adapters/capturefile"
15	14		"github.com/bnema/nefercap/internal/ports"
16	15		)
17	16

@@ -41,14 +40,11 @@ func (*Writer) Save(ctx context.Context, f ports.Frame, path string) (err error)
41	40		if path == "" {
42	41		return errors.New("png: empty path")
43	42		}
44		-	file, err := os.OpenFile(path, os.O_WRONLY\|os.O_CREATE\|os.O_EXCL, 0o600)
	43	+	file, res, err := capturefile.Create(path)
45	44		if err != nil {
46		-	if errors.Is(err, fs.ErrExist) {
47		-	return fmt.Errorf("%w: %s", ports.ErrPathExists, path)
48		-	}
49		-	return fmt.Errorf("png: create output: %w", err)
	45	+	return fmt.Errorf("png: %w", err)
50	46		}
51		-	// We created the file exclusively, so we own it and may remove it.
	47	+	// We created the file exclusively, so we own it; a replacement is not ours.
52	48		defer func() {
53	49		if err == nil {
54	50		if cerr := file.Close(); cerr != nil {

@@ -58,8 +54,8 @@ func (*Writer) Save(ctx context.Context, f ports.Frame, path string) (err error)
58	54		_ = file.Close()
59	55		}
60	56		if err != nil {
61		-	if rerr := os.Remove(path); rerr != nil && !errors.Is(rerr, fs.ErrNotExist) {
62		-	err = errors.Join(err, fmt.Errorf("png: remove partial output: %w", rerr))
	57	+	if rerr := res.Remove(); rerr != nil {
	58	+	err = errors.Join(err, fmt.Errorf("png: %w", rerr))
63	59		}
64	60		}
65	61		}()

@@ -101,7 +97,6 @@ func toRGBA(ctx context.Context, f ports.Frame) (*image.RGBA, error) {
101	97		}
102	98		src := f.Row(y)
103	99		dst := img.Pix[yrowLen : (y+1)rowLen]
104		-	dst = dst[:len(src)]
105	100		for x := 0; x+3 < len(src); x += 4 {
106	101		dst[x] = src[x+2]
107	102		dst[x+1] = src[x+1]

 	"image"
 	stdpng "image/png"
 	"io"
-	"io/fs"
-	"os"
+	"github.com/bnema/nefercap/internal/adapters/capturefile"
 	"github.com/bnema/nefercap/internal/ports"
 )
 	if path == "" {
 		return errors.New("png: empty path")
 	}
-	file, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
+	file, res, err := capturefile.Create(path)
 	if err != nil {
-		if errors.Is(err, fs.ErrExist) {
+		return fmt.Errorf("png: %w", err)
-			return fmt.Errorf("%w: %s", ports.ErrPathExists, path)
-		}
-		return fmt.Errorf("png: create output: %w", err)
 	}
-	// We created the file exclusively, so we own it and may remove it.
+	// We created the file exclusively, so we own it; a replacement is not ours.
 	defer func() {
 		if err == nil {
 			if cerr := file.Close(); cerr != nil {
 			_ = file.Close()
 		}
 		if err != nil {
-			if rerr := os.Remove(path); rerr != nil && !errors.Is(rerr, fs.ErrNotExist) {
+			if rerr := res.Remove(); rerr != nil {
-				err = errors.Join(err, fmt.Errorf("png: remove partial output: %w", rerr))
+				err = errors.Join(err, fmt.Errorf("png: %w", rerr))
 			}
 		}
 	}()
 		}
 		src := f.Row(y)
 		dst := img.Pix[y*rowLen : (y+1)*rowLen]
-		dst = dst[:len(src)]
 		for x := 0; x+3 < len(src); x += 4 {
 			dst[x] = src[x+2]
 			dst[x+1] = src[x+1]

M internal/adapters/png/png_test.go +34 -2


@@ -90,8 +90,8 @@ func TestSaveNeverOverwrites(t *testing.T) {
90	90		}
91	91		f := ports.Frame{Pixels: make([]byte, 4), Width: 1, Height: 1, Stride: 4, Format: ARGB}
92	92		err := New().Save(context.Background(), f, path)
93		-	if !errors.Is(err, ports.ErrPathExists) {
94		-	t.Fatalf("got %v, want ErrPathExists", err)
	93	+	if !errors.Is(err, ports.ErrPathExists) \|\| !errors.Is(err, os.ErrExist) {
	94	+	t.Fatalf("got %v, want ErrPathExists and fs.ErrExist", err)
95	95		}
96	96		if got, _ := os.ReadFile(path); string(got) != "keep" {
97	97		t.Fatalf("existing file changed: %q", got)

@@ -183,3 +183,35 @@ func BenchmarkSave(b *testing.B) {
183	183		})
184	184		}
185	185		}
	186	+
	187	+	func TestCancelCleanupSparesReplacementFile(t *testing.T) {
	188	+	const side = 2048
	189	+	pix := make([]byte, sideside4)
	190	+	rng := rand.New(rand.NewPCG(3, 4))
	191	+	for i := range pix {
	192	+	pix[i] = byte(rng.Uint32())
	193	+	}
	194	+	f := ports.Frame{Pixels: pix, Width: side, Height: side, Stride: side * 4, Format: ports.XRGB8888}
	195	+	dir := t.TempDir()
	196	+	path := filepath.Join(dir, "shot.png")
	197	+	moved := filepath.Join(dir, "moved.png")
	198	+	ctx, cancel := context.WithCancel(context.Background())
	199	+	defer cancel()
	200	+	go func() { // the user moves the file away and puts their own at the path
	201	+	for ctx.Err() == nil {
	202	+	if _, err := os.Stat(path); err == nil {
	203	+	if os.Rename(path, moved) == nil && os.WriteFile(path, []byte("user data"), 0o600) == nil {
	204	+	cancel()
	205	+	return
	206	+	}
	207	+	}
	208	+	time.Sleep(50 * time.Microsecond)
	209	+	}
	210	+	}()
	211	+	if err := New().Save(ctx, f, path); !errors.Is(err, context.Canceled) {
	212	+	t.Fatalf("got %v, want context.Canceled", err)
	213	+	}
	214	+	if got, err := os.ReadFile(path); err != nil \|\| string(got) != "user data" {
	215	+	t.Fatalf("replacement was removed or changed: %q, %v", got, err)
	216	+	}
	217	+	}


@@ -90,8 +90,8 @@ func TestSaveNeverOverwrites(t *testing.T) {
90		}	90		}
91		f := ports.Frame{Pixels: make([]byte, 4), Width: 1, Height: 1, Stride: 4, Format: ARGB}	91		f := ports.Frame{Pixels: make([]byte, 4), Width: 1, Height: 1, Stride: 4, Format: ARGB}
92		err := New().Save(context.Background(), f, path)	92		err := New().Save(context.Background(), f, path)
93	-	if !errors.Is(err, ports.ErrPathExists) {	93	+	if !errors.Is(err, ports.ErrPathExists) \|\| !errors.Is(err, os.ErrExist) {
94	-	t.Fatalf("got %v, want ErrPathExists", err)	94	+	t.Fatalf("got %v, want ErrPathExists and fs.ErrExist", err)
95		}	95		}
96		if got, _ := os.ReadFile(path); string(got) != "keep" {	96		if got, _ := os.ReadFile(path); string(got) != "keep" {
97		t.Fatalf("existing file changed: %q", got)	97		t.Fatalf("existing file changed: %q", got)

@@ -183,3 +183,35 @@ func BenchmarkSave(b *testing.B) {
183		})	183		})
184		}	184		}
185		}	185		}
			186	+
			187	+	func TestCancelCleanupSparesReplacementFile(t *testing.T) {
			188	+	const side = 2048
			189	+	pix := make([]byte, sideside4)
			190	+	rng := rand.New(rand.NewPCG(3, 4))
			191	+	for i := range pix {
			192	+	pix[i] = byte(rng.Uint32())
			193	+	}
			194	+	f := ports.Frame{Pixels: pix, Width: side, Height: side, Stride: side * 4, Format: ports.XRGB8888}
			195	+	dir := t.TempDir()
			196	+	path := filepath.Join(dir, "shot.png")
			197	+	moved := filepath.Join(dir, "moved.png")
			198	+	ctx, cancel := context.WithCancel(context.Background())
			199	+	defer cancel()
			200	+	go func() { // the user moves the file away and puts their own at the path
			201	+	for ctx.Err() == nil {
			202	+	if _, err := os.Stat(path); err == nil {
			203	+	if os.Rename(path, moved) == nil && os.WriteFile(path, []byte("user data"), 0o600) == nil {
			204	+	cancel()
			205	+	return
			206	+	}
			207	+	}
			208	+	time.Sleep(50 * time.Microsecond)
			209	+	}
			210	+	}()
			211	+	if err := New().Save(ctx, f, path); !errors.Is(err, context.Canceled) {
			212	+	t.Fatalf("got %v, want context.Canceled", err)
			213	+	}
			214	+	if got, err := os.ReadFile(path); err != nil \|\| string(got) != "user data" {
			215	+	t.Fatalf("replacement was removed or changed: %q, %v", got, err)
			216	+	}
			217	+	}