fix(inbox-worker): retry mute/favorite/read on a missing sub like roles

UpdateSubscriptionMute/Favorite/Read silently no-op'd when the subscription didn't
exist yet, while UpdateSubscriptionRoles Nak-retries. Because member_added runs on the
FIFO membership lane and the field events on the worker pool, a field event frequently
wins the race and finds no sub — silently dropping the migrated mute/favorite/unread
state. Add the same subscriptionExists disambiguation (extracted as naksIfSubscriptionMissing)
to all four: a genuinely missing sub now errors → redelivered until member_added lands,
while a stale event the high-water guard rejected stays a silent no-op. Follows the
review's "don't silently drop migrated state" principle (the field-state instance ngangwar962
flagged for status but not subscriptions). Tests flipped to assert the new Nak behavior.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012X9qhQT4NwmCHjdndwNtFD

Author: claude

inbox-worker/handler.go

@@ -37,19 +37,19 @@ type InboxStore interface {
 	// UpdateSubscriptionRead sets lastSeenAt and alert on the subscription
 	// keyed by (roomID, account). Idempotent and order-safe: the write
 	// only applies when the stored lastSeenAt is missing or strictly
-	// earlier than the supplied value. Older or duplicate events are
-	// silent no-ops. Missing-subscription is also a silent no-op.
+	// earlier than the supplied value. Older or duplicate events are silent no-ops.
+	// A genuinely missing sub returns an error (Nak) so the event redelivers until member_added lands.
 	UpdateSubscriptionRead(ctx context.Context, roomID, account string, lastSeenAt time.Time, alert bool) error
 	UpsertThreadSubscription(ctx context.Context, sub *model.ThreadSubscription) error
 	// ApplyThreadRead writes ThreadSubscription under a $lt lastSeenAt guard, then the Subscription only if the guard accepted.
 	ApplyThreadRead(ctx context.Context, roomID, threadRoomID, account string, newThreadUnread []string, alert bool, lastSeenAt time.Time) error
 	// UpdateSubscriptionMute sets muted by (roomID, account), guarded by
 	// muteUpdatedAt (the source event's publish time): older/duplicate events
-	// are silent no-ops. Missing-sub is also a silent no-op for federation races.
+	// are silent no-ops. A genuinely missing sub returns an error (Nak) so the event redelivers until member_added lands.
 	UpdateSubscriptionMute(ctx context.Context, roomID, account string, muted bool, muteUpdatedAt time.Time) error
 	// UpdateSubscriptionFavorite sets favorite by (roomID, account), guarded by
 	// favoriteUpdatedAt (the source event's publish time): older/duplicate events
-	// are silent no-ops. Missing-sub is also a silent no-op for federation races.
+	// are silent no-ops. A genuinely missing sub returns an error (Nak) so the event redelivers until member_added lands.
 	UpdateSubscriptionFavorite(ctx context.Context, roomID, account string, favorite bool, favoriteUpdatedAt time.Time) error
 	// UpdateSubscriptionNamesForRoom sets name on every subscription in the room,
 	// each guarded by its own nameUpdatedAt so an out-of-order rename cannot regress

inbox-worker/handler_test.go

@@ -1686,12 +1686,12 @@ func TestHandler_SubscriptionMuteToggled(t *testing.T) {
 	assert.Equal(t, int64(12345), updates[0].updatedAt.UnixMilli())
 }
 
-func TestHandler_SubscriptionMuteToggled_MissingSubscriptionNoOp(t *testing.T) {
+func TestHandler_SubscriptionMuteToggled_Forwarded(t *testing.T) {
 	store := &stubInboxStore{}
 	h := NewHandler(store)
 
 	payload, err := json.Marshal(model.SubscriptionMuteToggledEvent{
-		Account: "ghost", RoomID: "r1", Muted: true, Timestamp: 12345,
+		Account: "alice", RoomID: "r1", Muted: true, Timestamp: 12345,
 	})
 	require.NoError(t, err)
 	evt, err := json.Marshal(model.InboxEvent{
@@ -1701,6 +1701,10 @@ func TestHandler_SubscriptionMuteToggled_MissingSubscriptionNoOp(t *testing.T) {
 	require.NoError(t, err)
 
 	require.NoError(t, h.HandleEvent(context.Background(), evt))
+	updates := store.getMuteUpdates()
+	require.Len(t, updates, 1)
+	assert.Equal(t, "alice", updates[0].account)
+	assert.True(t, updates[0].muted)
 }
 
 func TestHandler_SubscriptionMuteToggled_MalformedPayload(t *testing.T) {
@@ -1751,12 +1755,12 @@ func TestHandler_SubscriptionFavoriteToggled(t *testing.T) {
 	assert.Equal(t, int64(12345), updates[0].updatedAt.UnixMilli())
 }
 
-func TestHandler_SubscriptionFavoriteToggled_MissingSubscriptionNoOp(t *testing.T) {
+func TestHandler_SubscriptionFavoriteToggled_Forwarded(t *testing.T) {
 	store := &stubInboxStore{}
 	h := NewHandler(store)
 
 	payload, err := json.Marshal(model.SubscriptionFavoriteToggledEvent{
-		Account: "ghost", RoomID: "r1", Favorite: true, Timestamp: 12345,
+		Account: "alice", RoomID: "r1", Favorite: true, Timestamp: 12345,
 	})
 	require.NoError(t, err)
 	evt, err := json.Marshal(model.InboxEvent{
@@ -1766,6 +1770,10 @@ func TestHandler_SubscriptionFavoriteToggled_MissingSubscriptionNoOp(t *testing.
 	require.NoError(t, err)
 
 	require.NoError(t, h.HandleEvent(context.Background(), evt))
+	updates := store.getFavoriteUpdates()
+	require.Len(t, updates, 1)
+	assert.Equal(t, "alice", updates[0].account)
+	assert.True(t, updates[0].favorite)
 }
 
 func TestHandler_SubscriptionFavoriteToggled_MalformedPayload(t *testing.T) {

inbox-worker/integration_test.go

@@ -337,18 +337,33 @@ func TestInbox_UpdateSubscriptionRead_EqualTimestampSkipped(t *testing.T) {
 	assert.True(t, got.Alert) // unchanged
 }
 
-func TestInbox_UpdateSubscriptionRead_MissingSubscriptionNoOp(t *testing.T) {
+func TestInbox_UpdateSubscriptionRead_MissingSubscriptionErrors(t *testing.T) {
 	ctx := context.Background()
-	db := setupMongo(t)
-	store := &mongoInboxStore{
-		subCol:       db.Collection("subscriptions"),
-		roomCol:      db.Collection("rooms"),
-		userCol:      db.Collection("users"),
-		threadSubCol: db.Collection("thread_subscriptions"),
-	}
+	store := newGuardStore(setupMongo(t))
+
+	// No subscription seeded — a genuinely missing sub must error so the event redelivers until
+	// member_added lands (field events can race ahead of member_added on the worker pool).
+	err := store.UpdateSubscriptionRead(ctx, "missing-room", "ghost", time.Now().UTC(), false)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "subscription not found")
+}
 
-	now := time.Now().UTC()
-	require.NoError(t, store.UpdateSubscriptionRead(ctx, "missing-room", "ghost", now, false))
+func TestInbox_UpdateSubscriptionMute_MissingSubscriptionErrors(t *testing.T) {
+	ctx := context.Background()
+	store := newGuardStore(setupMongo(t))
+
+	err := store.UpdateSubscriptionMute(ctx, "missing-room", "ghost", true, time.UnixMilli(100).UTC())
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "subscription not found")
+}
+
+func TestInbox_UpdateSubscriptionFavorite_MissingSubscriptionErrors(t *testing.T) {
+	ctx := context.Background()
+	store := newGuardStore(setupMongo(t))
+
+	err := store.UpdateSubscriptionFavorite(ctx, "missing-room", "ghost", true, time.UnixMilli(100).UTC())
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "subscription not found")
 }
 
 func TestInboxWorker_ThreadSubscriptionUpserted_Insert_Integration(t *testing.T) {

inbox-worker/main.go

@@ -105,9 +105,17 @@ func (s *mongoInboxStore) UpdateSubscriptionRoles(ctx context.Context, account,
 	if err != nil {
 		return fmt.Errorf("update subscription roles for %q in room %q: %w", account, roomID, err)
 	}
-	if res.MatchedCount > 0 {
-		return nil
+	if res.MatchedCount == 0 {
+		return s.naksIfSubscriptionMissing(ctx, account, roomID)
 	}
+	return nil
+}
+
+// naksIfSubscriptionMissing disambiguates a MatchedCount==0 guarded subscription write. A genuinely
+// missing subscription returns an error (Nak → redelivered until member_added lands, the
+// federation/migration race where field events can race ahead of member_added); a stale event the
+// high-water guard rejected is a silent no-op (the sub exists with a newer-or-equal value).
+func (s *mongoInboxStore) naksIfSubscriptionMissing(ctx context.Context, account, roomID string) error {
 	exists, err := s.subscriptionExists(ctx, account, roomID)
 	if err != nil {
 		return fmt.Errorf("check subscription exists for %q in room %q: %w", account, roomID, err)
@@ -236,9 +244,13 @@ func (s *mongoInboxStore) UpdateSubscriptionMute(ctx context.Context, roomID, ac
 		},
 	}
 	update := bson.M{"$set": bson.M{"muted": muted, "muteUpdatedAt": muteUpdatedAt}}
-	if _, err := s.subCol.UpdateOne(ctx, filter, update); err != nil {
+	res, err := s.subCol.UpdateOne(ctx, filter, update)
+	if err != nil {
 		return fmt.Errorf("update subscription mute for %q in room %q: %w", account, roomID, err)
 	}
+	if res.MatchedCount == 0 {
+		return s.naksIfSubscriptionMissing(ctx, account, roomID)
+	}
 	return nil
 }
 
@@ -256,9 +268,13 @@ func (s *mongoInboxStore) UpdateSubscriptionFavorite(ctx context.Context, roomID
 		},
 	}
 	update := bson.M{"$set": bson.M{"favorite": favorite, "favoriteUpdatedAt": favoriteUpdatedAt}}
-	if _, err := s.subCol.UpdateOne(ctx, filter, update); err != nil {
+	res, err := s.subCol.UpdateOne(ctx, filter, update)
+	if err != nil {
 		return fmt.Errorf("update subscription favorite for %q in room %q: %w", account, roomID, err)
 	}
+	if res.MatchedCount == 0 {
+		return s.naksIfSubscriptionMissing(ctx, account, roomID)
+	}
 	return nil
 }
 
@@ -272,9 +288,13 @@ func (s *mongoInboxStore) UpdateSubscriptionRead(ctx context.Context, roomID, ac
 		},
 	}
 	update := bson.M{"$set": bson.M{"lastSeenAt": lastSeenAt, "alert": alert}}
-	if _, err := s.subCol.UpdateOne(ctx, filter, update); err != nil {
+	res, err := s.subCol.UpdateOne(ctx, filter, update)
+	if err != nil {
 		return fmt.Errorf("update subscription read for %q in room %q: %w", account, roomID, err)
 	}
+	if res.MatchedCount == 0 {
+		return s.naksIfSubscriptionMissing(ctx, account, roomID)
+	}
 	return nil
 }