81c1931 fix(notices): serialize claim release on ack · vev

M internal/adapters/noticefile/noticefile.go +17 -8


@@ -40,10 +40,12 @@ type Store struct {
40	40		// production.
41	41		afterAppendOpen func()
42	42
43		-	// removeFile and syncDirectory make Ack cleanup failures deterministic in
44		-	// package tests. They are nil in production.
45		-	removeFile func(string) error
46		-	syncDirectory func(string) error
	43	+	// removeFile, syncDirectory, and beforeAckRelease make Ack cleanup and
	44	+	// claim-release ordering deterministic in package tests. They are nil in
	45	+	// production.
	46	+	removeFile func(string) error
	47	+	syncDirectory func(string) error
	48	+	beforeAckRelease func()
47	49		}
48	50
49	51		var (

@@ -329,14 +331,21 @@ func (s *Store) Ack() error {
329	331		if !s.ownsClaim() {
330	332		return ErrNoClaimOwner
331	333		}
332		-	// Ack is terminal for this Store's claim even when removing or syncing the
333		-	// record fails: retaining the flock would indefinitely block recovery.
334		-	defer s.releaseClaimLock()
335		-
336	334		return s.withLock(func() error {
337	335		if !s.ownsClaim() {
338	336		return ErrNoClaimOwner
339	337		}
	338	+	// Ack is terminal for this Store's claim even when removing or syncing
	339	+	// the record fails. Release while the operation lock is still held so a
	340	+	// concurrent Claim on this Store cannot see a removed in-flight file
	341	+	// while it still appears to own the claim.
	342	+	defer func() {
	343	+	if s.beforeAckRelease != nil {
	344	+	s.beforeAckRelease()
	345	+	}
	346	+	s.releaseClaimLock()
	347	+	}()
	348	+
340	349		remove := os.Remove
341	350		if s.removeFile != nil {
342	351		remove = s.removeFile

 	// production.
 	afterAppendOpen func()
-	// removeFile and syncDirectory make Ack cleanup failures deterministic in
+	// removeFile, syncDirectory, and beforeAckRelease make Ack cleanup and
-	// package tests. They are nil in production.
+	// claim-release ordering deterministic in package tests. They are nil in
-	removeFile    func(string) error
+	// production.
-	syncDirectory func(string) error
+	removeFile       func(string) error
+	syncDirectory    func(string) error
+	beforeAckRelease func()
 }
 var (
 	if !s.ownsClaim() {
 		return ErrNoClaimOwner
 	}
-	// Ack is terminal for this Store's claim even when removing or syncing the
-	// record fails: retaining the flock would indefinitely block recovery.
-	defer s.releaseClaimLock()
 	return s.withLock(func() error {
 		if !s.ownsClaim() {
 			return ErrNoClaimOwner
 		}
+		// Ack is terminal for this Store's claim even when removing or syncing
+		// the record fails. Release while the operation lock is still held so a
+		// concurrent Claim on this Store cannot see a removed in-flight file
+		// while it still appears to own the claim.
+		defer func() {
+			if s.beforeAckRelease != nil {
+				s.beforeAckRelease()
+			}
+			s.releaseClaimLock()
+		}()
 		remove := os.Remove
 		if s.removeFile != nil {
 			remove = s.removeFile

M internal/adapters/noticefile/noticefile_test.go +67 -0


@@ -6,6 +6,7 @@ import (
6	6		"os"
7	7		"path/filepath"
8	8		"strings"
	9	+	"syscall"
9	10		"testing"
10	11		"time"
11	12

@@ -157,6 +158,72 @@ func TestStoreAckReleasesClaimAfterCleanupFailure(t *testing.T) {
157	158		}
158	159		}
159	160
	161	+	func TestStoreAckSerializesClaimReleaseWithSameStoreReclaim(t *testing.T) {
	162	+	dir := filepath.Join(t.TempDir(), "vev")
	163	+	store := New(dir)
	164	+	require.NoError(t, store.Append(domain.Notification{Message: "owned"}))
	165	+	_, err := store.Claim()
	166	+	require.NoError(t, err)
	167	+
	168	+	releaseEntered := make(chan struct{})
	169	+	allowRelease := make(chan struct{})
	170	+	lockHeld := make(chan error, 1)
	171	+	store.beforeAckRelease = func() {
	172	+	held, err := operationLockHeld(store.lockPath())
	173	+	if err != nil {
	174	+	lockHeld <- err
	175	+	} else if !held {
	176	+	lockHeld <- errors.New("Ack released the operation lock before claim ownership")
	177	+	} else {
	178	+	lockHeld <- nil
	179	+	}
	180	+	close(releaseEntered)
	181	+	<-allowRelease
	182	+	}
	183	+
	184	+	ackDone := make(chan error, 1)
	185	+	go func() { ackDone <- store.Ack() }()
	186	+	<-releaseEntered // The in-flight file is gone, but Ack still owns the operation lock.
	187	+	require.NoError(t, <-lockHeld)
	188	+
	189	+	claimStarted := make(chan struct{})
	190	+	claimDone := make(chan struct {
	191	+	notices []domain.Notification
	192	+	err error
	193	+	}, 1)
	194	+	go func() {
	195	+	close(claimStarted)
	196	+	notices, err := store.Claim()
	197	+	claimDone <- struct {
	198	+	notices []domain.Notification
	199	+	err error
	200	+	}{notices, err}
	201	+	}()
	202	+	<-claimStarted
	203	+	close(allowRelease)
	204	+
	205	+	require.NoError(t, <-ackDone)
	206	+	reclaimed := <-claimDone
	207	+	require.NoError(t, reclaimed.err)
	208	+	require.Empty(t, reclaimed.notices)
	209	+	}
	210	+
	211	+	func operationLockHeld(path string) (bool, error) {
	212	+	f, err := os.OpenFile(path, os.O_CREATE\|os.O_RDWR, 0o600)
	213	+	if err != nil {
	214	+	return false, err
	215	+	}
	216	+	defer func() { _ = f.Close() }()
	217	+	if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX\|syscall.LOCK_NB); err != nil {
	218	+	if errors.Is(err, syscall.EWOULDBLOCK) {
	219	+	return true, nil
	220	+	}
	221	+	return false, err
	222	+	}
	223	+	defer func() { _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) }()
	224	+	return false, nil
	225	+	}
	226	+
160	227		func TestStoreClaimOwnershipPreventsLiveReplayOrAck(t *testing.T) {
161	228		t.Parallel()
162	229		dir := filepath.Join(t.TempDir(), "vev")


@@ -6,6 +6,7 @@ import (
6	"os"	6		"os"
7	"path/filepath"	7		"path/filepath"
8	"strings"	8		"strings"
		9	+	"syscall"
9	"testing"	10		"testing"
10	"time"	11		"time"
11		12

@@ -157,6 +158,72 @@ func TestStoreAckReleasesClaimAfterCleanupFailure(t *testing.T) {
157	}	158		}
158	}	159		}
159		160
		161	+	func TestStoreAckSerializesClaimReleaseWithSameStoreReclaim(t *testing.T) {
		162	+	dir := filepath.Join(t.TempDir(), "vev")
		163	+	store := New(dir)
		164	+	require.NoError(t, store.Append(domain.Notification{Message: "owned"}))
		165	+	_, err := store.Claim()
		166	+	require.NoError(t, err)
		167	+
		168	+	releaseEntered := make(chan struct{})
		169	+	allowRelease := make(chan struct{})
		170	+	lockHeld := make(chan error, 1)
		171	+	store.beforeAckRelease = func() {
		172	+	held, err := operationLockHeld(store.lockPath())
		173	+	if err != nil {
		174	+	lockHeld <- err
		175	+	} else if !held {
		176	+	lockHeld <- errors.New("Ack released the operation lock before claim ownership")
		177	+	} else {
		178	+	lockHeld <- nil
		179	+	}
		180	+	close(releaseEntered)
		181	+	<-allowRelease
		182	+	}
		183	+
		184	+	ackDone := make(chan error, 1)
		185	+	go func() { ackDone <- store.Ack() }()
		186	+	<-releaseEntered // The in-flight file is gone, but Ack still owns the operation lock.
		187	+	require.NoError(t, <-lockHeld)
		188	+
		189	+	claimStarted := make(chan struct{})
		190	+	claimDone := make(chan struct {
		191	+	notices []domain.Notification
		192	+	err error
		193	+	}, 1)
		194	+	go func() {
		195	+	close(claimStarted)
		196	+	notices, err := store.Claim()
		197	+	claimDone <- struct {
		198	+	notices []domain.Notification
		199	+	err error
		200	+	}{notices, err}
		201	+	}()
		202	+	<-claimStarted
		203	+	close(allowRelease)
		204	+
		205	+	require.NoError(t, <-ackDone)
		206	+	reclaimed := <-claimDone
		207	+	require.NoError(t, reclaimed.err)
		208	+	require.Empty(t, reclaimed.notices)
		209	+	}
		210	+
		211	+	func operationLockHeld(path string) (bool, error) {
		212	+	f, err := os.OpenFile(path, os.O_CREATE\|os.O_RDWR, 0o600)
		213	+	if err != nil {
		214	+	return false, err
		215	+	}
		216	+	defer func() { _ = f.Close() }()
		217	+	if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX\|syscall.LOCK_NB); err != nil {
		218	+	if errors.Is(err, syscall.EWOULDBLOCK) {
		219	+	return true, nil
		220	+	}
		221	+	return false, err
		222	+	}
		223	+	defer func() { _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) }()
		224	+	return false, nil
		225	+	}
		226	+
160	func TestStoreClaimOwnershipPreventsLiveReplayOrAck(t *testing.T) {	227		func TestStoreClaimOwnershipPreventsLiveReplayOrAck(t *testing.T) {
161	t.Parallel()	228		t.Parallel()
162	dir := filepath.Join(t.TempDir(), "vev")	229		dir := filepath.Join(t.TempDir(), "vev")

M internal/usecase/daemon/notices_overlay.go +1 -0


@@ -96,6 +96,7 @@ func handleListInputLocked(clock ports.Clock, data []byte, state listInputState,
96	96		continue
97	97		}
98	98		result.exit = true
	99	+	return result
99	100		default:
100	101		custom := action(data[i])
101	102		if custom.action != 0 {


@@ -96,6 +96,7 @@ func handleListInputLocked(clock ports.Clock, data []byte, state listInputState,
96	continue	96		continue
97	}	97		}
98	result.exit = true	98		result.exit = true
		99	+	return result
99	default:	100		default:
100	custom := action(data[i])	101		custom := action(data[i])
101	if custom.action != 0 {	102		if custom.action != 0 {

M internal/usecase/daemon/notices_overlay_test.go +8 -0


@@ -244,6 +244,14 @@ func TestHandleListInputStopsAfterExitInBatchedInput(t *testing.T) {
244	244		custom: func(b byte) listInputResult { return listInputResult{action: b} },
245	245		wantExit: true,
246	246		},
	247	+	{
	248	+	name: "invalid escape exit ignores trailing navigation and actions",
	249	+	input: []byte{'\x1b', 'x', 'j', 'y'},
	250	+	custom: func(b byte) listInputResult { return listInputResult{action: b} },
	251	+	wantExit: true,
	252	+	wantCalls: nil,
	253	+	wantAction: 0,
	254	+	},
247	255		{
248	256		name: "custom exit retains action and ignores trailing bytes",
249	257		input: []byte("xjy"),


@@ -244,6 +244,14 @@ func TestHandleListInputStopsAfterExitInBatchedInput(t *testing.T) {
244	custom: func(b byte) listInputResult { return listInputResult{action: b} },	244		custom: func(b byte) listInputResult { return listInputResult{action: b} },
245	wantExit: true,	245		wantExit: true,
246	},	246		},
		247	+	{
		248	+	name: "invalid escape exit ignores trailing navigation and actions",
		249	+	input: []byte{'\x1b', 'x', 'j', 'y'},
		250	+	custom: func(b byte) listInputResult { return listInputResult{action: b} },
		251	+	wantExit: true,
		252	+	wantCalls: nil,
		253	+	wantAction: 0,
		254	+	},
247	{	255		{
248	name: "custom exit retains action and ignores trailing bytes",	256		name: "custom exit retains action and ignores trailing bytes",
249	input: []byte("xjy"),	257		input: []byte("xjy"),