fix(noticefile): make pending notice import recoverable

commit: 6cda4a5725f43631e3aaebe58a1dc61d071b3c04
author: brice · 3mo ago · 2026-07-21 10:34 UTC
parent: b0ecd73
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +291 -133

M internal/adapters/noticefile/noticefile.go +131 -25


@@ -10,17 +10,27 @@ import (
10	10		"fmt"
11	11		"os"
12	12		"path/filepath"
	13	+	"syscall"
13	14
14	15		"github.com/bnema/vev/internal/domain"
15	16		"github.com/bnema/vev/internal/ports"
16	17		"github.com/bnema/vev/pkg/safedir"
17	18		)
18	19
19		-	const fileName = "pending-notices.jsonl"
	20	+	const (
	21	+	fileName = "pending-notices.jsonl"
	22	+	inFlightFileName = "pending-notices.inflight.jsonl"
	23	+	lockFileName = "pending-notices.lock"
	24	+	)
20	25
21	26		// Store persists notices as JSON-lines under dir.
22	27		type Store struct {
23	28		dir string
	29	+
	30	+	// afterClaimRead is used by package tests to pause a claim after its reader
	31	+	// reached EOF while it still owns the interprocess lock. It is nil in
	32	+	// production.
	33	+	afterClaimRead func()
24	34		}
25	35
26	36		var _ ports.NoticeStore = (*Store)(nil)

@@ -34,38 +44,126 @@ func (s *Store) path() string {
34	44		return filepath.Join(s.dir, fileName)
35	45		}
36	46
37		-	// Append writes n as one JSON line, creating the private state dir if needed.
38		-	func (s *Store) Append(n domain.Notification) error {
	47	+	func (s *Store) inFlightPath() string {
	48	+	return filepath.Join(s.dir, inFlightFileName)
	49	+	}
	50	+
	51	+	func (s *Store) lockPath() string {
	52	+	return filepath.Join(s.dir, lockFileName)
	53	+	}
	54	+
	55	+	// withLock serializes Append, Claim, and Ack across Store instances and
	56	+	// processes. Holding this lock while Claim rotates and reads the file prevents
	57	+	// an appender with an already-open file descriptor from writing into a claim
	58	+	// after its reader reached EOF.
	59	+	func (s *Store) withLock(fn func() error) error {
39	60		if err := safedir.EnsurePrivate(s.dir); err != nil {
40	61		return fmt.Errorf("create notice dir: %w", err)
41	62		}
42		-	f, err := os.OpenFile(s.path(), os.O_APPEND\|os.O_CREATE\|os.O_WRONLY, 0o600)
	63	+	f, err := os.OpenFile(s.lockPath(), os.O_CREATE\|os.O_RDWR, 0o600)
43	64		if err != nil {
44		-	return fmt.Errorf("open pending notices: %w", err)
	65	+	return fmt.Errorf("open notice lock: %w", err)
45	66		}
46	67		defer func() { _ = f.Close() }()
	68	+	if err := f.Chmod(0o600); err != nil {
	69	+	return fmt.Errorf("secure notice lock: %w", err)
	70	+	}
	71	+	if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX); err != nil {
	72	+	return fmt.Errorf("lock pending notices: %w", err)
	73	+	}
	74	+	defer func() { _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) }()
	75	+	return fn()
	76	+	}
47	77
	78	+	func syncDir(dir string) error {
	79	+	f, err := os.Open(dir)
	80	+	if err != nil {
	81	+	return err
	82	+	}
	83	+	defer func() { _ = f.Close() }()
	84	+	return f.Sync()
	85	+	}
	86	+
	87	+	// Append writes n as one JSON line, creating the private state dir if needed.
	88	+	// A successful append is synced before it is made available to a later claim.
	89	+	func (s *Store) Append(n domain.Notification) error {
48	90		data, err := json.Marshal(n)
49	91		if err != nil {
50	92		return fmt.Errorf("marshal notice: %w", err)
51	93		}
52	94		data = append(data, '\n')
53		-	if _, err := f.Write(data); err != nil {
54		-	return fmt.Errorf("write notice: %w", err)
55		-	}
56		-	return nil
	95	+
	96	+	return s.withLock(func() error {
	97	+	f, err := os.OpenFile(s.path(), os.O_APPEND\|os.O_CREATE\|os.O_WRONLY, 0o600)
	98	+	if err != nil {
	99	+	return fmt.Errorf("open pending notices: %w", err)
	100	+	}
	101	+	if err := f.Chmod(0o600); err != nil {
	102	+	_ = f.Close()
	103	+	return fmt.Errorf("secure pending notices: %w", err)
	104	+	}
	105	+	if _, err := f.Write(data); err != nil {
	106	+	_ = f.Close()
	107	+	return fmt.Errorf("write notice: %w", err)
	108	+	}
	109	+	if err := f.Sync(); err != nil {
	110	+	_ = f.Close()
	111	+	return fmt.Errorf("sync pending notices: %w", err)
	112	+	}
	113	+	if err := f.Close(); err != nil {
	114	+	return fmt.Errorf("close pending notices: %w", err)
	115	+	}
	116	+	if err := syncDir(s.dir); err != nil {
	117	+	return fmt.Errorf("sync notice dir: %w", err)
	118	+	}
	119	+	return nil
	120	+	})
57	121		}
58	122
59		-	// Drain returns all stored notices and truncates the store. A missing file
60		-	// means a fresh daemon has nothing pending, not an error.
61		-	func (s *Store) Drain() ([]domain.Notification, error) {
62		-	f, err := os.Open(s.path())
63		-	if errors.Is(err, os.ErrNotExist) {
64		-	return nil, nil
	123	+	// Claim atomically moves pending notices into an in-flight file and returns
	124	+	// their valid JSONL entries in order. The in-flight file is intentionally kept
	125	+	// until Ack, so an unacknowledged import is replayed after a crash. A missing
	126	+	// pending file means there is nothing to claim.
	127	+	func (s *Store) Claim() ([]domain.Notification, error) {
	128	+	var notices []domain.Notification
	129	+	err := s.withLock(func() error {
	130	+	inFlight := s.inFlightPath()
	131	+	if _, err := os.Stat(inFlight); err != nil {
	132	+	if !errors.Is(err, os.ErrNotExist) {
	133	+	return fmt.Errorf("stat in-flight notices: %w", err)
	134	+	}
	135	+	if err := os.Rename(s.path(), inFlight); err != nil {
	136	+	if errors.Is(err, os.ErrNotExist) {
	137	+	return nil
	138	+	}
	139	+	return fmt.Errorf("claim pending notices: %w", err)
	140	+	}
	141	+	// fsync the parent directory so the atomic claim is recoverable after
	142	+	// power loss, not merely after a clean process restart.
	143	+	if err := syncDir(s.dir); err != nil {
	144	+	return fmt.Errorf("sync claimed notices: %w", err)
	145	+	}
	146	+	}
	147	+
	148	+	var err error
	149	+	notices, err = readNotices(inFlight)
	150	+	if err == nil && s.afterClaimRead != nil {
	151	+	s.afterClaimRead()
	152	+	}
	153	+	return err
	154	+	})
	155	+	if err != nil {
	156	+	return nil, err
65	157		}
	158	+	return notices, nil
	159	+	}
	160	+
	161	+	func readNotices(path string) ([]domain.Notification, error) {
	162	+	f, err := os.Open(path)
66	163		if err != nil {
67		-	return nil, fmt.Errorf("open pending notices: %w", err)
	164	+	return nil, fmt.Errorf("open claimed notices: %w", err)
68	165		}
	166	+	defer func() { _ = f.Close() }()
69	167
70	168		var notices []domain.Notification
71	169		scanner := bufio.NewScanner(f)

@@ -77,18 +175,26 @@ func (s *Store) Drain() ([]domain.Notification, error) {
77	175		}
78	176		var n domain.Notification
79	177		if err := json.Unmarshal(line, &n); err != nil {
80		-	continue // corrupt or partial line must not abort the drain
	178	+	continue // corrupt or partial line must not abort the claim
81	179		}
82	180		notices = append(notices, n)
83	181		}
84		-	scanErr := scanner.Err()
85		-	_ = f.Close()
86		-	if scanErr != nil {
87		-	return nil, fmt.Errorf("read pending notices: %w", scanErr)
88		-	}
89		-
90		-	if err := os.Truncate(s.path(), 0); err != nil {
91		-	return nil, fmt.Errorf("truncate pending notices: %w", err)
	182	+	if err := scanner.Err(); err != nil {
	183	+	return nil, fmt.Errorf("read claimed notices: %w", err)
92	184		}
93	185		return notices, nil
94	186		}
	187	+
	188	+	// Ack permanently removes the current in-flight claim. It is only called once
	189	+	// the daemon has recorded and queued every claimed notice.
	190	+	func (s *Store) Ack() error {
	191	+	return s.withLock(func() error {
	192	+	if err := os.Remove(s.inFlightPath()); err != nil && !errors.Is(err, os.ErrNotExist) {
	193	+	return fmt.Errorf("ack claimed notices: %w", err)
	194	+	}
	195	+	if err := syncDir(s.dir); err != nil {
	196	+	return fmt.Errorf("sync acknowledged notices: %w", err)
	197	+	}
	198	+	return nil
	199	+	})
	200	+	}


@@ -10,17 +10,27 @@ import (
10		"fmt"	10		"fmt"
11		"os"	11		"os"
12		"path/filepath"	12		"path/filepath"
			13	+	"syscall"
13			14
14		"github.com/bnema/vev/internal/domain"	15		"github.com/bnema/vev/internal/domain"
15		"github.com/bnema/vev/internal/ports"	16		"github.com/bnema/vev/internal/ports"
16		"github.com/bnema/vev/pkg/safedir"	17		"github.com/bnema/vev/pkg/safedir"
17		)	18		)
18			19
19	-	const fileName = "pending-notices.jsonl"	20	+	const (
			21	+	fileName = "pending-notices.jsonl"
			22	+	inFlightFileName = "pending-notices.inflight.jsonl"
			23	+	lockFileName = "pending-notices.lock"
			24	+	)
20			25
21		// Store persists notices as JSON-lines under dir.	26		// Store persists notices as JSON-lines under dir.
22		type Store struct {	27		type Store struct {
23		dir string	28		dir string
			29	+
			30	+	// afterClaimRead is used by package tests to pause a claim after its reader
			31	+	// reached EOF while it still owns the interprocess lock. It is nil in
			32	+	// production.
			33	+	afterClaimRead func()
24		}	34		}
25			35
26		var _ ports.NoticeStore = (*Store)(nil)	36		var _ ports.NoticeStore = (*Store)(nil)

@@ -34,38 +44,126 @@ func (s *Store) path() string {
34		return filepath.Join(s.dir, fileName)	44		return filepath.Join(s.dir, fileName)
35		}	45		}
36			46
37	-	// Append writes n as one JSON line, creating the private state dir if needed.	47	+	func (s *Store) inFlightPath() string {
38	-	func (s *Store) Append(n domain.Notification) error {	48	+	return filepath.Join(s.dir, inFlightFileName)
			49	+	}
			50	+
			51	+	func (s *Store) lockPath() string {
			52	+	return filepath.Join(s.dir, lockFileName)
			53	+	}
			54	+
			55	+	// withLock serializes Append, Claim, and Ack across Store instances and
			56	+	// processes. Holding this lock while Claim rotates and reads the file prevents
			57	+	// an appender with an already-open file descriptor from writing into a claim
			58	+	// after its reader reached EOF.
			59	+	func (s *Store) withLock(fn func() error) error {
39		if err := safedir.EnsurePrivate(s.dir); err != nil {	60		if err := safedir.EnsurePrivate(s.dir); err != nil {
40		return fmt.Errorf("create notice dir: %w", err)	61		return fmt.Errorf("create notice dir: %w", err)
41		}	62		}
42	-	f, err := os.OpenFile(s.path(), os.O_APPEND\|os.O_CREATE\|os.O_WRONLY, 0o600)	63	+	f, err := os.OpenFile(s.lockPath(), os.O_CREATE\|os.O_RDWR, 0o600)
43		if err != nil {	64		if err != nil {
44	-	return fmt.Errorf("open pending notices: %w", err)	65	+	return fmt.Errorf("open notice lock: %w", err)
45		}	66		}
46		defer func() { _ = f.Close() }()	67		defer func() { _ = f.Close() }()
			68	+	if err := f.Chmod(0o600); err != nil {
			69	+	return fmt.Errorf("secure notice lock: %w", err)
			70	+	}
			71	+	if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX); err != nil {
			72	+	return fmt.Errorf("lock pending notices: %w", err)
			73	+	}
			74	+	defer func() { _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) }()
			75	+	return fn()
			76	+	}
47			77
			78	+	func syncDir(dir string) error {
			79	+	f, err := os.Open(dir)
			80	+	if err != nil {
			81	+	return err
			82	+	}
			83	+	defer func() { _ = f.Close() }()
			84	+	return f.Sync()
			85	+	}
			86	+
			87	+	// Append writes n as one JSON line, creating the private state dir if needed.
			88	+	// A successful append is synced before it is made available to a later claim.
			89	+	func (s *Store) Append(n domain.Notification) error {
48		data, err := json.Marshal(n)	90		data, err := json.Marshal(n)
49		if err != nil {	91		if err != nil {
50		return fmt.Errorf("marshal notice: %w", err)	92		return fmt.Errorf("marshal notice: %w", err)
51		}	93		}
52		data = append(data, '\n')	94		data = append(data, '\n')
53	-	if _, err := f.Write(data); err != nil {	95	+
54	-	return fmt.Errorf("write notice: %w", err)	96	+	return s.withLock(func() error {
55	-	}	97	+	f, err := os.OpenFile(s.path(), os.O_APPEND\|os.O_CREATE\|os.O_WRONLY, 0o600)
56	-	return nil	98	+	if err != nil {
			99	+	return fmt.Errorf("open pending notices: %w", err)
			100	+	}
			101	+	if err := f.Chmod(0o600); err != nil {
			102	+	_ = f.Close()
			103	+	return fmt.Errorf("secure pending notices: %w", err)
			104	+	}
			105	+	if _, err := f.Write(data); err != nil {
			106	+	_ = f.Close()
			107	+	return fmt.Errorf("write notice: %w", err)
			108	+	}
			109	+	if err := f.Sync(); err != nil {
			110	+	_ = f.Close()
			111	+	return fmt.Errorf("sync pending notices: %w", err)
			112	+	}
			113	+	if err := f.Close(); err != nil {
			114	+	return fmt.Errorf("close pending notices: %w", err)
			115	+	}
			116	+	if err := syncDir(s.dir); err != nil {
			117	+	return fmt.Errorf("sync notice dir: %w", err)
			118	+	}
			119	+	return nil
			120	+	})
57		}	121		}
58			122
59	-	// Drain returns all stored notices and truncates the store. A missing file	123	+	// Claim atomically moves pending notices into an in-flight file and returns
60	-	// means a fresh daemon has nothing pending, not an error.	124	+	// their valid JSONL entries in order. The in-flight file is intentionally kept
61	-	func (s *Store) Drain() ([]domain.Notification, error) {	125	+	// until Ack, so an unacknowledged import is replayed after a crash. A missing
62	-	f, err := os.Open(s.path())	126	+	// pending file means there is nothing to claim.
63	-	if errors.Is(err, os.ErrNotExist) {	127	+	func (s *Store) Claim() ([]domain.Notification, error) {
64	-	return nil, nil	128	+	var notices []domain.Notification
			129	+	err := s.withLock(func() error {
			130	+	inFlight := s.inFlightPath()
			131	+	if _, err := os.Stat(inFlight); err != nil {
			132	+	if !errors.Is(err, os.ErrNotExist) {
			133	+	return fmt.Errorf("stat in-flight notices: %w", err)
			134	+	}
			135	+	if err := os.Rename(s.path(), inFlight); err != nil {
			136	+	if errors.Is(err, os.ErrNotExist) {
			137	+	return nil
			138	+	}
			139	+	return fmt.Errorf("claim pending notices: %w", err)
			140	+	}
			141	+	// fsync the parent directory so the atomic claim is recoverable after
			142	+	// power loss, not merely after a clean process restart.
			143	+	if err := syncDir(s.dir); err != nil {
			144	+	return fmt.Errorf("sync claimed notices: %w", err)
			145	+	}
			146	+	}
			147	+
			148	+	var err error
			149	+	notices, err = readNotices(inFlight)
			150	+	if err == nil && s.afterClaimRead != nil {
			151	+	s.afterClaimRead()
			152	+	}
			153	+	return err
			154	+	})
			155	+	if err != nil {
			156	+	return nil, err
65		}	157		}
			158	+	return notices, nil
			159	+	}
			160	+
			161	+	func readNotices(path string) ([]domain.Notification, error) {
			162	+	f, err := os.Open(path)
66		if err != nil {	163		if err != nil {
67	-	return nil, fmt.Errorf("open pending notices: %w", err)	164	+	return nil, fmt.Errorf("open claimed notices: %w", err)
68		}	165		}
			166	+	defer func() { _ = f.Close() }()
69			167
70		var notices []domain.Notification	168		var notices []domain.Notification
71		scanner := bufio.NewScanner(f)	169		scanner := bufio.NewScanner(f)

@@ -77,18 +175,26 @@ func (s *Store) Drain() ([]domain.Notification, error) {
77		}	175		}
78		var n domain.Notification	176		var n domain.Notification
79		if err := json.Unmarshal(line, &n); err != nil {	177		if err := json.Unmarshal(line, &n); err != nil {
80	-	continue // corrupt or partial line must not abort the drain	178	+	continue // corrupt or partial line must not abort the claim
81		}	179		}
82		notices = append(notices, n)	180		notices = append(notices, n)
83		}	181		}
84	-	scanErr := scanner.Err()	182	+	if err := scanner.Err(); err != nil {
85	-	_ = f.Close()	183	+	return nil, fmt.Errorf("read claimed notices: %w", err)
86	-	if scanErr != nil {
87	-	return nil, fmt.Errorf("read pending notices: %w", scanErr)
88	-	}
89	-
90	-	if err := os.Truncate(s.path(), 0); err != nil {
91	-	return nil, fmt.Errorf("truncate pending notices: %w", err)
92		}	184		}
93		return notices, nil	185		return notices, nil
94		}	186		}
			187	+
			188	+	// Ack permanently removes the current in-flight claim. It is only called once
			189	+	// the daemon has recorded and queued every claimed notice.
			190	+	func (s *Store) Ack() error {
			191	+	return s.withLock(func() error {
			192	+	if err := os.Remove(s.inFlightPath()); err != nil && !errors.Is(err, os.ErrNotExist) {
			193	+	return fmt.Errorf("ack claimed notices: %w", err)
			194	+	}
			195	+	if err := syncDir(s.dir); err != nil {
			196	+	return fmt.Errorf("sync acknowledged notices: %w", err)
			197	+	}
			198	+	return nil
			199	+	})
			200	+	}

M internal/adapters/noticefile/noticefile_test.go +81 -83


@@ -6,10 +6,12 @@ import (
6	6		"testing"
7	7		"time"
8	8
	9	+	"github.com/stretchr/testify/require"
	10	+
9	11		"github.com/bnema/vev/internal/domain"
10	12		)
11	13
12		-	func TestStoreAppendDrainRoundTrip(t *testing.T) {
	14	+	func TestStoreClaimAckRoundTrip(t *testing.T) {
13	15		t.Parallel()
14	16		dir := filepath.Join(t.TempDir(), "vev")
15	17		store := New(dir)

@@ -30,110 +32,106 @@ func TestStoreAppendDrainRoundTrip(t *testing.T) {
30	32		Time: time.Date(2026, 7, 20, 10, 5, 0, 0, time.UTC),
31	33		Count: 2,
32	34		}
	35	+	require.NoError(t, store.Append(first))
	36	+	require.NoError(t, store.Append(second))
33	37
34		-	if err := store.Append(first); err != nil {
35		-	t.Fatalf("Append first: %v", err)
36		-	}
37		-	if err := store.Append(second); err != nil {
38		-	t.Fatalf("Append second: %v", err)
39		-	}
	38	+	claimed, err := store.Claim()
	39	+	require.NoError(t, err)
	40	+	require.Equal(t, []domain.Notification{first, second}, claimed)
40	41
41		-	got, err := store.Drain()
42		-	if err != nil {
43		-	t.Fatalf("Drain: %v", err)
44		-	}
45		-	if len(got) != 2 {
46		-	t.Fatalf("Drain len = %d, want 2", len(got))
47		-	}
48		-	if !got[0].Time.Equal(first.Time) \|\| got[0].Message != first.Message \|\| got[0].Code != first.Code \|\|
49		-	got[0].Severity != first.Severity \|\| got[0].Details != first.Details \|\| got[0].Count != first.Count \|\|
50		-	got[0].SessionID != first.SessionID {
51		-	t.Fatalf("first notice round-trip mismatch: got %#v, want %#v", got[0], first)
52		-	}
53		-	if !got[1].Time.Equal(second.Time) \|\| got[1].Message != second.Message \|\| got[1].Code != second.Code \|\|
54		-	got[1].Severity != second.Severity \|\| got[1].Count != second.Count {
55		-	t.Fatalf("second notice round-trip mismatch: got %#v, want %#v", got[1], second)
56		-	}
	42	+	// A fresh store simulates a daemon crash after Claim but before Ack.
	43	+	recovered := New(dir)
	44	+	replayed, err := recovered.Claim()
	45	+	require.NoError(t, err)
	46	+	require.Equal(t, claimed, replayed)
57	47
58		-	again, err := store.Drain()
59		-	if err != nil {
60		-	t.Fatalf("Drain again: %v", err)
61		-	}
62		-	if len(again) != 0 {
63		-	t.Fatalf("Drain again len = %d, want 0", len(again))
64		-	}
	48	+	require.NoError(t, recovered.Ack())
	49	+	again, err := New(dir).Claim()
	50	+	require.NoError(t, err)
	51	+	require.Empty(t, again)
65	52		}
66	53
67		-	func TestStoreDrainMissingFileReturnsNil(t *testing.T) {
	54	+	func TestStoreClaimDoesNotLoseAppendAfterRotation(t *testing.T) {
68	55		t.Parallel()
69		-	dir := t.TempDir()
	56	+	dir := filepath.Join(t.TempDir(), "vev")
70	57		store := New(dir)
	58	+	first := domain.Notification{Message: "before claim"}
	59	+	second := domain.Notification{Message: "after claim rotation"}
	60	+	require.NoError(t, store.Append(first))
71	61
72		-	got, err := store.Drain()
73		-	if err != nil {
74		-	t.Fatalf("Drain: %v", err)
75		-	}
76		-	if got != nil {
77		-	t.Fatalf("Drain on missing file = %#v, want nil", got)
78		-	}
	62	+	rotated := make(chan struct{})
	63	+	allowRead := make(chan struct{})
	64	+	store.afterClaimRead = func() {
	65	+	close(rotated)
	66	+	<-allowRead
	67	+	}
	68	+	claimDone := make(chan []domain.Notification, 1)
	69	+	errDone := make(chan error, 1)
	70	+	go func() {
	71	+	claimed, err := store.Claim()
	72	+	claimDone <- claimed
	73	+	errDone <- err
	74	+	}()
	75	+	<-rotated
	76	+
	77	+	appendDone := make(chan error, 1)
	78	+	go func() { appendDone <- New(dir).Append(second) }()
	79	+	close(allowRead)
	80	+	require.NoError(t, <-errDone)
	81	+	require.Equal(t, []domain.Notification{first}, <-claimDone)
	82	+	require.NoError(t, <-appendDone)
	83	+
	84	+	require.NoError(t, store.Ack())
	85	+	claimed, err := New(dir).Claim()
	86	+	require.NoError(t, err)
	87	+	require.Equal(t, []domain.Notification{second}, claimed)
	88	+	require.NoError(t, New(dir).Ack())
79	89		}
80	90
81		-	func TestStoreDrainSkipsGarbageLines(t *testing.T) {
	91	+	func TestStoreClaimMissingFileReturnsNil(t *testing.T) {
82	92		t.Parallel()
83		-	dir := t.TempDir()
	93	+
	94	+	got, err := New(filepath.Join(t.TempDir(), "vev")).Claim()
	95	+	require.NoError(t, err)
	96	+	require.Nil(t, got)
	97	+	}
	98	+
	99	+	func TestStoreClaimSkipsGarbageLinesInOrder(t *testing.T) {
	100	+	t.Parallel()
	101	+	dir := filepath.Join(t.TempDir(), "vev")
84	102		store := New(dir)
85		-	path := filepath.Join(dir, "pending-notices.jsonl")
86		-	if err := os.MkdirAll(dir, 0o700); err != nil {
87		-	t.Fatalf("MkdirAll: %v", err)
88		-	}
	103	+	path := filepath.Join(dir, fileName)
	104	+	require.NoError(t, os.MkdirAll(dir, 0o700))
89	105
90	106		good := `{"Code":9,"Severity":2,"Message":"first"}` + "\n"
91	107		garbage := "not json at all\n"
92	108		good2 := `{"Code":10,"Severity":1,"Message":"second"}` + "\n"
93		-	if err := os.WriteFile(path, []byte(good+garbage+good2), 0o600); err != nil {
94		-	t.Fatalf("WriteFile: %v", err)
95		-	}
96		-
97		-	got, err := store.Drain()
98		-	if err != nil {
99		-	t.Fatalf("Drain: %v", err)
100		-	}
101		-	if len(got) != 2 {
102		-	t.Fatalf("Drain len = %d, want 2 (garbage line skipped)", len(got))
103		-	}
104		-	if got[0].Message != "first" \|\| got[1].Message != "second" {
105		-	t.Fatalf("Drain messages = %q, %q; want first, second", got[0].Message, got[1].Message)
106		-	}
	109	+	require.NoError(t, os.WriteFile(path, []byte(good+garbage+good2), 0o600))
107	110
108		-	data, err := os.ReadFile(path)
109		-	if err != nil {
110		-	t.Fatalf("ReadFile after drain: %v", err)
111		-	}
112		-	if len(data) != 0 {
113		-	t.Fatalf("file not truncated after Drain, got %d bytes", len(data))
114		-	}
	111	+	got, err := store.Claim()
	112	+	require.NoError(t, err)
	113	+	require.Len(t, got, 2)
	114	+	require.Equal(t, "first", got[0].Message)
	115	+	require.Equal(t, "second", got[1].Message)
	116	+	require.NoError(t, store.Ack())
115	117		}
116	118
117		-	func TestStoreAppendPrivatizesDir(t *testing.T) {
	119	+	func TestStoreAppendUsesPrivatePermissions(t *testing.T) {
118	120		t.Parallel()
119	121		dir := filepath.Join(t.TempDir(), "vev")
120	122		store := New(dir)
121	123
122		-	if err := store.Append(domain.Notification{Message: "x"}); err != nil {
123		-	t.Fatalf("Append: %v", err)
124		-	}
	124	+	require.NoError(t, store.Append(domain.Notification{Message: "x"}))
125	125		info, err := os.Stat(dir)
126		-	if err != nil {
127		-	t.Fatalf("Stat dir: %v", err)
128		-	}
129		-	if got := info.Mode().Perm(); got != 0o700 {
130		-	t.Fatalf("dir mode = %v, want 0700", got)
131		-	}
132		-	fi, err := os.Stat(filepath.Join(dir, "pending-notices.jsonl"))
133		-	if err != nil {
134		-	t.Fatalf("Stat file: %v", err)
135		-	}
136		-	if got := fi.Mode().Perm(); got != 0o600 {
137		-	t.Fatalf("file mode = %v, want 0600", got)
138		-	}
	126	+	require.NoError(t, err)
	127	+	require.Equal(t, os.FileMode(0o700), info.Mode().Perm())
	128	+	fi, err := os.Stat(filepath.Join(dir, fileName))
	129	+	require.NoError(t, err)
	130	+	require.Equal(t, os.FileMode(0o600), fi.Mode().Perm())
	131	+
	132	+	_, err = store.Claim()
	133	+	require.NoError(t, err)
	134	+	fi, err = os.Stat(filepath.Join(dir, inFlightFileName))
	135	+	require.NoError(t, err)
	136	+	require.Equal(t, os.FileMode(0o600), fi.Mode().Perm())
139	137		}


@@ -6,10 +6,12 @@ import (
6		"testing"	6		"testing"
7		"time"	7		"time"
8			8
			9	+	"github.com/stretchr/testify/require"
			10	+
9		"github.com/bnema/vev/internal/domain"	11		"github.com/bnema/vev/internal/domain"
10		)	12		)
11			13
12	-	func TestStoreAppendDrainRoundTrip(t *testing.T) {	14	+	func TestStoreClaimAckRoundTrip(t *testing.T) {
13		t.Parallel()	15		t.Parallel()
14		dir := filepath.Join(t.TempDir(), "vev")	16		dir := filepath.Join(t.TempDir(), "vev")
15		store := New(dir)	17		store := New(dir)

@@ -30,110 +32,106 @@ func TestStoreAppendDrainRoundTrip(t *testing.T) {
30		Time: time.Date(2026, 7, 20, 10, 5, 0, 0, time.UTC),	32		Time: time.Date(2026, 7, 20, 10, 5, 0, 0, time.UTC),
31		Count: 2,	33		Count: 2,
32		}	34		}
			35	+	require.NoError(t, store.Append(first))
			36	+	require.NoError(t, store.Append(second))
33			37
34	-	if err := store.Append(first); err != nil {	38	+	claimed, err := store.Claim()
35	-	t.Fatalf("Append first: %v", err)	39	+	require.NoError(t, err)
36	-	}	40	+	require.Equal(t, []domain.Notification{first, second}, claimed)
37	-	if err := store.Append(second); err != nil {
38	-	t.Fatalf("Append second: %v", err)
39	-	}
40			41
41	-	got, err := store.Drain()	42	+	// A fresh store simulates a daemon crash after Claim but before Ack.
42	-	if err != nil {	43	+	recovered := New(dir)
43	-	t.Fatalf("Drain: %v", err)	44	+	replayed, err := recovered.Claim()
44	-	}	45	+	require.NoError(t, err)
45	-	if len(got) != 2 {	46	+	require.Equal(t, claimed, replayed)
46	-	t.Fatalf("Drain len = %d, want 2", len(got))
47	-	}
48	-	if !got[0].Time.Equal(first.Time) \|\| got[0].Message != first.Message \|\| got[0].Code != first.Code \|\|
49	-	got[0].Severity != first.Severity \|\| got[0].Details != first.Details \|\| got[0].Count != first.Count \|\|
50	-	got[0].SessionID != first.SessionID {
51	-	t.Fatalf("first notice round-trip mismatch: got %#v, want %#v", got[0], first)
52	-	}
53	-	if !got[1].Time.Equal(second.Time) \|\| got[1].Message != second.Message \|\| got[1].Code != second.Code \|\|
54	-	got[1].Severity != second.Severity \|\| got[1].Count != second.Count {
55	-	t.Fatalf("second notice round-trip mismatch: got %#v, want %#v", got[1], second)
56	-	}
57			47
58	-	again, err := store.Drain()	48	+	require.NoError(t, recovered.Ack())
59	-	if err != nil {	49	+	again, err := New(dir).Claim()
60	-	t.Fatalf("Drain again: %v", err)	50	+	require.NoError(t, err)
61	-	}	51	+	require.Empty(t, again)
62	-	if len(again) != 0 {
63	-	t.Fatalf("Drain again len = %d, want 0", len(again))
64	-	}
65		}	52		}
66			53
67	-	func TestStoreDrainMissingFileReturnsNil(t *testing.T) {	54	+	func TestStoreClaimDoesNotLoseAppendAfterRotation(t *testing.T) {
68		t.Parallel()	55		t.Parallel()
69	-	dir := t.TempDir()	56	+	dir := filepath.Join(t.TempDir(), "vev")
70		store := New(dir)	57		store := New(dir)
			58	+	first := domain.Notification{Message: "before claim"}
			59	+	second := domain.Notification{Message: "after claim rotation"}
			60	+	require.NoError(t, store.Append(first))
71			61
72	-	got, err := store.Drain()	62	+	rotated := make(chan struct{})
73	-	if err != nil {	63	+	allowRead := make(chan struct{})
74	-	t.Fatalf("Drain: %v", err)	64	+	store.afterClaimRead = func() {
75	-	}	65	+	close(rotated)
76	-	if got != nil {	66	+	<-allowRead
77	-	t.Fatalf("Drain on missing file = %#v, want nil", got)	67	+	}
78	-	}	68	+	claimDone := make(chan []domain.Notification, 1)
			69	+	errDone := make(chan error, 1)
			70	+	go func() {
			71	+	claimed, err := store.Claim()
			72	+	claimDone <- claimed
			73	+	errDone <- err
			74	+	}()
			75	+	<-rotated
			76	+
			77	+	appendDone := make(chan error, 1)
			78	+	go func() { appendDone <- New(dir).Append(second) }()
			79	+	close(allowRead)
			80	+	require.NoError(t, <-errDone)
			81	+	require.Equal(t, []domain.Notification{first}, <-claimDone)
			82	+	require.NoError(t, <-appendDone)
			83	+
			84	+	require.NoError(t, store.Ack())
			85	+	claimed, err := New(dir).Claim()
			86	+	require.NoError(t, err)
			87	+	require.Equal(t, []domain.Notification{second}, claimed)
			88	+	require.NoError(t, New(dir).Ack())
79		}	89		}
80			90
81	-	func TestStoreDrainSkipsGarbageLines(t *testing.T) {	91	+	func TestStoreClaimMissingFileReturnsNil(t *testing.T) {
82		t.Parallel()	92		t.Parallel()
83	-	dir := t.TempDir()	93	+
			94	+	got, err := New(filepath.Join(t.TempDir(), "vev")).Claim()
			95	+	require.NoError(t, err)
			96	+	require.Nil(t, got)
			97	+	}
			98	+
			99	+	func TestStoreClaimSkipsGarbageLinesInOrder(t *testing.T) {
			100	+	t.Parallel()
			101	+	dir := filepath.Join(t.TempDir(), "vev")
84		store := New(dir)	102		store := New(dir)
85	-	path := filepath.Join(dir, "pending-notices.jsonl")	103	+	path := filepath.Join(dir, fileName)
86	-	if err := os.MkdirAll(dir, 0o700); err != nil {	104	+	require.NoError(t, os.MkdirAll(dir, 0o700))
87	-	t.Fatalf("MkdirAll: %v", err)
88	-	}
89			105
90		good := `{"Code":9,"Severity":2,"Message":"first"}` + "\n"	106		good := `{"Code":9,"Severity":2,"Message":"first"}` + "\n"
91		garbage := "not json at all\n"	107		garbage := "not json at all\n"
92		good2 := `{"Code":10,"Severity":1,"Message":"second"}` + "\n"	108		good2 := `{"Code":10,"Severity":1,"Message":"second"}` + "\n"
93	-	if err := os.WriteFile(path, []byte(good+garbage+good2), 0o600); err != nil {	109	+	require.NoError(t, os.WriteFile(path, []byte(good+garbage+good2), 0o600))
94	-	t.Fatalf("WriteFile: %v", err)
95	-	}
96	-
97	-	got, err := store.Drain()
98	-	if err != nil {
99	-	t.Fatalf("Drain: %v", err)
100	-	}
101	-	if len(got) != 2 {
102	-	t.Fatalf("Drain len = %d, want 2 (garbage line skipped)", len(got))
103	-	}
104	-	if got[0].Message != "first" \|\| got[1].Message != "second" {
105	-	t.Fatalf("Drain messages = %q, %q; want first, second", got[0].Message, got[1].Message)
106	-	}
107			110
108	-	data, err := os.ReadFile(path)	111	+	got, err := store.Claim()
109	-	if err != nil {	112	+	require.NoError(t, err)
110	-	t.Fatalf("ReadFile after drain: %v", err)	113	+	require.Len(t, got, 2)
111	-	}	114	+	require.Equal(t, "first", got[0].Message)
112	-	if len(data) != 0 {	115	+	require.Equal(t, "second", got[1].Message)
113	-	t.Fatalf("file not truncated after Drain, got %d bytes", len(data))	116	+	require.NoError(t, store.Ack())
114	-	}
115		}	117		}
116			118
117	-	func TestStoreAppendPrivatizesDir(t *testing.T) {	119	+	func TestStoreAppendUsesPrivatePermissions(t *testing.T) {
118		t.Parallel()	120		t.Parallel()
119		dir := filepath.Join(t.TempDir(), "vev")	121		dir := filepath.Join(t.TempDir(), "vev")
120		store := New(dir)	122		store := New(dir)
121			123
122	-	if err := store.Append(domain.Notification{Message: "x"}); err != nil {	124	+	require.NoError(t, store.Append(domain.Notification{Message: "x"}))
123	-	t.Fatalf("Append: %v", err)
124	-	}
125		info, err := os.Stat(dir)	125		info, err := os.Stat(dir)
126	-	if err != nil {	126	+	require.NoError(t, err)
127	-	t.Fatalf("Stat dir: %v", err)	127	+	require.Equal(t, os.FileMode(0o700), info.Mode().Perm())
128	-	}	128	+	fi, err := os.Stat(filepath.Join(dir, fileName))
129	-	if got := info.Mode().Perm(); got != 0o700 {	129	+	require.NoError(t, err)
130	-	t.Fatalf("dir mode = %v, want 0700", got)	130	+	require.Equal(t, os.FileMode(0o600), fi.Mode().Perm())
131	-	}	131	+
132	-	fi, err := os.Stat(filepath.Join(dir, "pending-notices.jsonl"))	132	+	_, err = store.Claim()
133	-	if err != nil {	133	+	require.NoError(t, err)
134	-	t.Fatalf("Stat file: %v", err)	134	+	fi, err = os.Stat(filepath.Join(dir, inFlightFileName))
135	-	}	135	+	require.NoError(t, err)
136	-	if got := fi.Mode().Perm(); got != 0o600 {	136	+	require.Equal(t, os.FileMode(0o600), fi.Mode().Perm())
137	-	t.Fatalf("file mode = %v, want 0600", got)
138	-	}
139		}	137		}

M internal/ports/mocks/notice_mock.go +55 -11


@@ -36,6 +36,50 @@ func (_m MockNoticeStore) EXPECT() MockNoticeStore_Expecter {
36	36		return &MockNoticeStore_Expecter{mock: &_m.Mock}
37	37		}
38	38
	39	+	// Ack provides a mock function for the type MockNoticeStore
	40	+	func (_mock *MockNoticeStore) Ack() error {
	41	+	ret := _mock.Called()
	42	+
	43	+	if len(ret) == 0 {
	44	+	panic("no return value specified for Ack")
	45	+	}
	46	+
	47	+	var r0 error
	48	+	if returnFunc, ok := ret.Get(0).(func() error); ok {
	49	+	r0 = returnFunc()
	50	+	} else {
	51	+	r0 = ret.Error(0)
	52	+	}
	53	+	return r0
	54	+	}
	55	+
	56	+	// MockNoticeStore_Ack_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Ack'
	57	+	type MockNoticeStore_Ack_Call struct {
	58	+	*mock.Call
	59	+	}
	60	+
	61	+	// Ack is a helper method to define mock.On call
	62	+	func (_e MockNoticeStore_Expecter) Ack() MockNoticeStore_Ack_Call {
	63	+	return &MockNoticeStore_Ack_Call{Call: _e.mock.On("Ack")}
	64	+	}
	65	+
	66	+	func (_c MockNoticeStore_Ack_Call) Run(run func()) MockNoticeStore_Ack_Call {
	67	+	_c.Call.Run(func(args mock.Arguments) {
	68	+	run()
	69	+	})
	70	+	return _c
	71	+	}
	72	+
	73	+	func (_c MockNoticeStore_Ack_Call) Return(err error) MockNoticeStore_Ack_Call {
	74	+	_c.Call.Return(err)
	75	+	return _c
	76	+	}
	77	+
	78	+	func (_c MockNoticeStore_Ack_Call) RunAndReturn(run func() error) MockNoticeStore_Ack_Call {
	79	+	_c.Call.Return(run)
	80	+	return _c
	81	+	}
	82	+
39	83		// Append provides a mock function for the type MockNoticeStore
40	84		func (_mock *MockNoticeStore) Append(n domain.Notification) error {
41	85		ret := _mock.Called(n)

@@ -87,12 +131,12 @@ func (_c *MockNoticeStore_Append_Call) RunAndReturn(run func(n domain.Notificati
87	131		return _c
88	132		}
89	133
90		-	// Drain provides a mock function for the type MockNoticeStore
91		-	func (_mock *MockNoticeStore) Drain() ([]domain.Notification, error) {
	134	+	// Claim provides a mock function for the type MockNoticeStore
	135	+	func (_mock *MockNoticeStore) Claim() ([]domain.Notification, error) {
92	136		ret := _mock.Called()
93	137
94	138		if len(ret) == 0 {
95		-	panic("no return value specified for Drain")
	139	+	panic("no return value specified for Claim")
96	140		}
97	141
98	142		var r0 []domain.Notification

@@ -115,29 +159,29 @@ func (_mock *MockNoticeStore) Drain() ([]domain.Notification, error) {
115	159		return r0, r1
116	160		}
117	161
118		-	// MockNoticeStore_Drain_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Drain'
119		-	type MockNoticeStore_Drain_Call struct {
	162	+	// MockNoticeStore_Claim_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Claim'
	163	+	type MockNoticeStore_Claim_Call struct {
120	164		*mock.Call
121	165		}
122	166
123		-	// Drain is a helper method to define mock.On call
124		-	func (_e MockNoticeStore_Expecter) Drain() MockNoticeStore_Drain_Call {
125		-	return &MockNoticeStore_Drain_Call{Call: _e.mock.On("Drain")}
	167	+	// Claim is a helper method to define mock.On call
	168	+	func (_e MockNoticeStore_Expecter) Claim() MockNoticeStore_Claim_Call {
	169	+	return &MockNoticeStore_Claim_Call{Call: _e.mock.On("Claim")}
126	170		}
127	171
128		-	func (_c MockNoticeStore_Drain_Call) Run(run func()) MockNoticeStore_Drain_Call {
	172	+	func (_c MockNoticeStore_Claim_Call) Run(run func()) MockNoticeStore_Claim_Call {
129	173		_c.Call.Run(func(args mock.Arguments) {
130	174		run()
131	175		})
132	176		return _c
133	177		}
134	178
135		-	func (_c MockNoticeStore_Drain_Call) Return(notifications []domain.Notification, err error) MockNoticeStore_Drain_Call {
	179	+	func (_c MockNoticeStore_Claim_Call) Return(notifications []domain.Notification, err error) MockNoticeStore_Claim_Call {
136	180		_c.Call.Return(notifications, err)
137	181		return _c
138	182		}
139	183
140		-	func (_c MockNoticeStore_Drain_Call) RunAndReturn(run func() ([]domain.Notification, error)) MockNoticeStore_Drain_Call {
	184	+	func (_c MockNoticeStore_Claim_Call) RunAndReturn(run func() ([]domain.Notification, error)) MockNoticeStore_Claim_Call {
141	185		_c.Call.Return(run)
142	186		return _c
143	187		}

 	return &MockNoticeStore_Expecter{mock: &_m.Mock}
 }
+// Ack provides a mock function for the type MockNoticeStore
+func (_mock *MockNoticeStore) Ack() error {
+	ret := _mock.Called()
+	if len(ret) == 0 {
+		panic("no return value specified for Ack")
+	}
+	var r0 error
+	if returnFunc, ok := ret.Get(0).(func() error); ok {
+		r0 = returnFunc()
+	} else {
+		r0 = ret.Error(0)
+	}
+	return r0
+}
+// MockNoticeStore_Ack_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Ack'
+type MockNoticeStore_Ack_Call struct {
+	*mock.Call
+}
+// Ack is a helper method to define mock.On call
+func (_e *MockNoticeStore_Expecter) Ack() *MockNoticeStore_Ack_Call {
+	return &MockNoticeStore_Ack_Call{Call: _e.mock.On("Ack")}
+}
+func (_c *MockNoticeStore_Ack_Call) Run(run func()) *MockNoticeStore_Ack_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run()
+	})
+	return _c
+}
+func (_c *MockNoticeStore_Ack_Call) Return(err error) *MockNoticeStore_Ack_Call {
+	_c.Call.Return(err)
+	return _c
+}
+func (_c *MockNoticeStore_Ack_Call) RunAndReturn(run func() error) *MockNoticeStore_Ack_Call {
+	_c.Call.Return(run)
+	return _c
+}
 // Append provides a mock function for the type MockNoticeStore
 func (_mock *MockNoticeStore) Append(n domain.Notification) error {
 	ret := _mock.Called(n)
 	return _c
 }
-// Drain provides a mock function for the type MockNoticeStore
+// Claim provides a mock function for the type MockNoticeStore
-func (_mock *MockNoticeStore) Drain() ([]domain.Notification, error) {
+func (_mock *MockNoticeStore) Claim() ([]domain.Notification, error) {
 	ret := _mock.Called()
 	if len(ret) == 0 {
-		panic("no return value specified for Drain")
+		panic("no return value specified for Claim")
 	}
 	var r0 []domain.Notification
 	return r0, r1
 }
-// MockNoticeStore_Drain_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Drain'
+// MockNoticeStore_Claim_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Claim'
-type MockNoticeStore_Drain_Call struct {
+type MockNoticeStore_Claim_Call struct {
 	*mock.Call
 }
-// Drain is a helper method to define mock.On call
+// Claim is a helper method to define mock.On call
-func (_e *MockNoticeStore_Expecter) Drain() *MockNoticeStore_Drain_Call {
+func (_e *MockNoticeStore_Expecter) Claim() *MockNoticeStore_Claim_Call {
-	return &MockNoticeStore_Drain_Call{Call: _e.mock.On("Drain")}
+	return &MockNoticeStore_Claim_Call{Call: _e.mock.On("Claim")}
 }
-func (_c *MockNoticeStore_Drain_Call) Run(run func()) *MockNoticeStore_Drain_Call {
+func (_c *MockNoticeStore_Claim_Call) Run(run func()) *MockNoticeStore_Claim_Call {
 	_c.Call.Run(func(args mock.Arguments) {
 		run()
 	})
 	return _c
 }
-func (_c *MockNoticeStore_Drain_Call) Return(notifications []domain.Notification, err error) *MockNoticeStore_Drain_Call {
+func (_c *MockNoticeStore_Claim_Call) Return(notifications []domain.Notification, err error) *MockNoticeStore_Claim_Call {
 	_c.Call.Return(notifications, err)
 	return _c
 }
-func (_c *MockNoticeStore_Drain_Call) RunAndReturn(run func() ([]domain.Notification, error)) *MockNoticeStore_Drain_Call {
+func (_c *MockNoticeStore_Claim_Call) RunAndReturn(run func() ([]domain.Notification, error)) *MockNoticeStore_Claim_Call {
 	_c.Call.Return(run)
 	return _c
 }

M internal/ports/notice.go +5 -2


@@ -5,6 +5,9 @@ import "github.com/bnema/vev/internal/domain"
5	5		// NoticeStore persists undeliverable notices across daemon restarts.
6	6		type NoticeStore interface {
7	7		Append(n domain.Notification) error
8		-	// Drain returns all stored notices and truncates the store.
9		-	Drain() ([]domain.Notification, error)
	8	+	// Claim atomically reserves pending notices for import. A claim remains
	9	+	// recoverable until Ack is called, including across process crashes.
	10	+	Claim() ([]domain.Notification, error)
	11	+	// Ack permanently removes the notices returned by the current claim.
	12	+	Ack() error
10	13		}

 // NoticeStore persists undeliverable notices across daemon restarts.
 type NoticeStore interface {
 	Append(n domain.Notification) error
-	// Drain returns all stored notices and truncates the store.
+	// Claim atomically reserves pending notices for import. A claim remains
-	Drain() ([]domain.Notification, error)
+	// recoverable until Ack is called, including across process crashes.
+	Claim() ([]domain.Notification, error)
+	// Ack permanently removes the notices returned by the current claim.
+	Ack() error
 }

M internal/usecase/daemon/notice_persist_test.go +7 -6


@@ -103,16 +103,17 @@ func TestKillSessionServerShutdownNotSaturatedSkipsAppend(t *testing.T) {
103	103		require.NoError(t, err)
104	104		}
105	105
106		-	// TestStartupDrainRecordsAndToastsPendingNotices proves notices persisted by a
107		-	// previous daemon are recorded to history and toasted to the first client that
108		-	// attaches after this daemon starts.
109		-	func TestStartupDrainRecordsAndToastsPendingNotices(t *testing.T) {
110		-	drained := []domain.Notification{
	106	+	// TestStartupClaimRecordsAndToastsPendingNotices proves notices persisted by
	107	+	// a previous daemon are recorded to history and toasted to the first client
	108	+	// that attaches after this daemon starts before their claim is acknowledged.
	109	+	func TestStartupClaimRecordsAndToastsPendingNotices(t *testing.T) {
	110	+	claimed := []domain.Notification{
111	111		{Code: domain.NoticeSnapshotWrite, Severity: domain.NoticeError, Message: "session a shut down without saving terminal state"},
112	112		{Code: domain.NoticeSnapshotWrite, Severity: domain.NoticeError, Message: "session b shut down without saving terminal state"},
113	113		}
114	114		notices := portsmocks.NewMockNoticeStore(t)
115		-	notices.EXPECT().Drain().Return(drained, nil).Once()
	115	+	notices.EXPECT().Claim().Return(claimed, nil).Once()
	116	+	notices.EXPECT().Ack().Return(nil).Once()
116	117
117	118		p, releasePTY := newBlockingPTY(t)
118	119		defer releasePTY()

 	require.NoError(t, err)
 }
-// TestStartupDrainRecordsAndToastsPendingNotices proves notices persisted by a
+// TestStartupClaimRecordsAndToastsPendingNotices proves notices persisted by
-// previous daemon are recorded to history and toasted to the first client that
+// a previous daemon are recorded to history and toasted to the first client
-// attaches after this daemon starts.
+// that attaches after this daemon starts before their claim is acknowledged.
-func TestStartupDrainRecordsAndToastsPendingNotices(t *testing.T) {
+func TestStartupClaimRecordsAndToastsPendingNotices(t *testing.T) {
-	drained := []domain.Notification{
+	claimed := []domain.Notification{
 		{Code: domain.NoticeSnapshotWrite, Severity: domain.NoticeError, Message: "session a shut down without saving terminal state"},
 		{Code: domain.NoticeSnapshotWrite, Severity: domain.NoticeError, Message: "session b shut down without saving terminal state"},
 	}
 	notices := portsmocks.NewMockNoticeStore(t)
-	notices.EXPECT().Drain().Return(drained, nil).Once()
+	notices.EXPECT().Claim().Return(claimed, nil).Once()
+	notices.EXPECT().Ack().Return(nil).Once()
 	p, releasePTY := newBlockingPTY(t)
 	defer releasePTY()

M internal/usecase/daemon/snapshot.go +12 -6


@@ -67,13 +67,19 @@ func (d *Daemon) restoreSnapshots(ctx context.Context) {
67	67		return
68	68		}
69	69		if ns := d.noticeStore; ns != nil {
70		-	drained, err := ns.Drain()
	70	+	claimed, err := ns.Claim()
71	71		if err != nil {
72		-	d.log.Warn("draining pending notices failed", "err", err)
73		-	}
74		-	for _, n := range drained {
75		-	d.notices.record(n)
76		-	d.notices.queueGlobal(n)
	72	+	d.log.Warn("claiming pending notices failed", "err", err)
	73	+	} else {
	74	+	for _, n := range claimed {
	75	+	d.notices.record(n)
	76	+	d.notices.queueGlobal(n)
	77	+	}
	78	+	// Do not discard the claim until every notice has been recorded and
	79	+	// queued. An Ack failure is safe: the claim is replayed at startup.
	80	+	if err := ns.Ack(); err != nil {
	81	+	d.log.Warn("acknowledging pending notices failed", "err", err)
	82	+	}
77	83		}
78	84		}
79	85		if !d.snapsEnabled \|\| d.snaps == nil {

 		return
 	}
 	if ns := d.noticeStore; ns != nil {
-		drained, err := ns.Drain()
+		claimed, err := ns.Claim()
 		if err != nil {
-			d.log.Warn("draining pending notices failed", "err", err)
+			d.log.Warn("claiming pending notices failed", "err", err)
-		}
+		} else {
-		for _, n := range drained {
+			for _, n := range claimed {
-			d.notices.record(n)
+				d.notices.record(n)
-			d.notices.queueGlobal(n)
+				d.notices.queueGlobal(n)
+			}
+			// Do not discard the claim until every notice has been recorded and
+			// queued. An Ack failure is safe: the claim is replayed at startup.
+			if err := ns.Ack(); err != nil {
+				d.log.Warn("acknowledging pending notices failed", "err", err)
+			}
 		}
 	}
 	if !d.snapsEnabled || d.snaps == nil {