enhance(websocket): improve client synchronization (#38847)

Keep notification counts and stopwatches in sync when tabs open or
reconnect, closing the gap behind [this
flake](https://github.com/go-gitea/gitea/actions/runs/31312525323/job/93242429394).

1. Send current notification counts and stopwatch state on each
WebSocket connection, replacing the separate state fetches tabs made on
every connect.
2. Tabs joining an already open connection fetch current state once,
since its snapshot went out before they attached.
3. Send events queued during the handshake before the snapshot, so older
events can't roll it back.
4. Wait for the Redis SUBSCRIBE acknowledgement, so the snapshot read
after subscribing can't miss an event.

routers/web/websocket/websocket.go

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	"gitea.dev/modules/graceful"
-	"gitea.dev/modules/json"
 	"gitea.dev/modules/log"
 	"gitea.dev/services/context"
 	"gitea.dev/services/pubsub"
@@ -30,23 +29,6 @@ const (
 	closeCodeUnauthenticated gitea_ws.StatusCode = 3000
 )
 
-// filterLogout forwards a session-free logout only to the targeted connection
-// (its own session, or every session when SessionID is empty) and drops it for
-// the rest. Non-logout messages pass through untouched.
-func filterLogout(eventType string, eventDataBytes []byte, connSessionID string) []byte {
-	if eventType != websocket_service.EventLogout {
-		return eventDataBytes
-	}
-	var lm websocket_service.UserEventMessage[websocket_service.LogoutEventData]
-	if err := json.Unmarshal(eventDataBytes, &lm); err != nil {
-		return eventDataBytes
-	}
-	if lm.EventData.SessionID == "" || lm.EventData.SessionID == connSessionID {
-		return []byte(`{"eventType":"logout"}`)
-	}
-	return nil
-}
-
 func Serve(ctx *context.Context) {
 	// Answer plain GETs (health checks, crawlers) here; letting Accept reject them
 	// would log an error per request. Same reply it would have sent.
@@ -57,6 +39,14 @@ func Serve(ctx *context.Context) {
 		return
 	}
 
+	if !ctx.IsSigned {
+		rejectUnauthenticated(ctx)
+		return
+	}
+
+	ch, cancel := pubsub.DefaultBroker.Subscribe(pubsub.UserTopic(ctx.Doer.ID)) // before the snapshot, so no change falls between them
+	defer cancel()
+
 	conn, err := gitea_ws.Accept(ctx.Resp, ctx.Req, nil)
 	if err != nil {
 		log.Error("websocket: accept failed: %v", err)
@@ -64,14 +54,7 @@ func Serve(ctx *context.Context) {
 	}
 	defer conn.CloseNow() //nolint:errcheck // best-effort close
 
-	if !ctx.IsSigned {
-		_ = conn.Close(closeCodeUnauthenticated, "unauthenticated")
-		return
-	}
-
 	sessionID := ctx.Session.ID()
-	ch, cancel := pubsub.DefaultBroker.Subscribe(pubsub.UserTopic(ctx.Doer.ID))
-	defer cancel()
 
 	// Ping requires a concurrent reader to observe the pong frame; CloseRead
 	// spawns one and cancels its context when the peer goes away.
@@ -80,6 +63,30 @@ func Serve(ctx *context.Context) {
 	pingTicker := time.NewTicker(pingInterval)
 	defer pingTicker.Stop()
 
+	send := func(brokerPayload []byte) error {
+		eventDataBytes := websocket_service.EventForSession(brokerPayload, sessionID)
+		if eventDataBytes == nil {
+			return nil
+		}
+		writeCtx, cancelWrite := gocontext.WithTimeout(wsCtx, writeTimeout) // a stalled peer must not starve the ping ticker
+		defer cancelWrite()
+		err := conn.Write(writeCtx, gitea_ws.MessageText, eventDataBytes)
+		if err != nil {
+			log.Trace("websocket: write failed: %v", err)
+		}
+		return err
+	}
+
+	var backlog [][]byte // published before the snapshot read, so it must not follow the snapshot
+	for len(ch) > 0 {
+		backlog = append(backlog, <-ch)
+	}
+	for _, brokerPayload := range append(backlog, websocket_service.UserData(wsCtx, ctx.Doer)...) {
+		if err := send(brokerPayload); err != nil {
+			return
+		}
+	}
+
 	for {
 		select {
 		case <-wsCtx.Done():
@@ -98,20 +105,19 @@ func Serve(ctx *context.Context) {
 			if !ok {
 				return
 			}
-			eventType, eventDataBytes := websocket_service.ExtractUserEventMessage(brokerPayload)
-			eventDataBytes = filterLogout(eventType, eventDataBytes, sessionID)
-			if eventDataBytes == nil {
-				continue
-			}
-			// Bound the write so a stalled/slow peer can't block this goroutine
-			// indefinitely and starve the ping ticker.
-			writeCtx, cancelWrite := gocontext.WithTimeout(wsCtx, writeTimeout)
-			err := conn.Write(writeCtx, gitea_ws.MessageText, eventDataBytes)
-			cancelWrite()
-			if err != nil {
-				log.Trace("websocket: write failed: %v", err)
+			if err := send(brokerPayload); err != nil {
 				return
 			}
 		}
 	}
 }
+
+// rejectUnauthenticated upgrades only to close, so clients stop reconnecting.
+func rejectUnauthenticated(ctx *context.Context) {
+	conn, err := gitea_ws.Accept(ctx.Resp, ctx.Req, nil)
+	if err != nil {
+		log.Error("websocket: accept failed: %v", err)
+		return
+	}
+	_ = conn.Close(closeCodeUnauthenticated, "unauthenticated")
+}

routers/web/websocket/websocket_test.go

@@ -1,60 +0,0 @@
-// Copyright 2026 The Gitea Authors. All rights reserved.
-// SPDX-License-Identifier: MIT
-
-package websocket
-
-import (
-	"testing"
-
-	"gitea.dev/services/websocket"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestFilterLogout(t *testing.T) {
-	cases := []struct {
-		name       string
-		brokerMsg  []byte
-		connSessID string
-		want       []byte // expected payload forwarded to the client
-	}{
-		{
-			name:       "originating session gets a session-free logout",
-			brokerMsg:  websocket.MakeUserEventMessage("logout", websocket.LogoutEventData{SessionID: "sess-A"}),
-			connSessID: "sess-A",
-			want:       []byte(`{"eventType":"logout"}`),
-		},
-		{
-			name:       "other session is dropped",
-			brokerMsg:  websocket.MakeUserEventMessage("logout", websocket.LogoutEventData{SessionID: "sess-A"}),
-			connSessID: "sess-B",
-			want:       nil,
-		},
-		{
-			name:       "empty sessionID reaches every session",
-			brokerMsg:  websocket.MakeUserEventMessage("logout", nil),
-			connSessID: "sess-A",
-			want:       []byte(`{"eventType":"logout"}`),
-		},
-		{
-			name:       "non-logout message passes through unchanged",
-			brokerMsg:  websocket.MakeUserEventMessage("other", map[string]any{"k": "v"}),
-			connSessID: "sess-A",
-			want:       []byte(`{"eventType":"other","eventData":{"k":"v"}}`),
-		},
-		{
-			name:       "malformed JSON with logout marker passes through unchanged",
-			brokerMsg:  []byte("any type\nany data"),
-			connSessID: "sess-A",
-			want:       []byte("any data"),
-		},
-	}
-
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			eventType, eventData := websocket.ExtractUserEventMessage(tc.brokerMsg)
-			out := filterLogout(eventType, eventData, tc.connSessID)
-			assert.Equal(t, tc.want, out)
-		})
-	}
-}

services/pubsub/broker_test.go

@@ -29,7 +29,6 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur
 		b := newBroker(t)
 		ch, cancel := b.Subscribe(t.Name())
 		defer cancel()
-		waitSubscribed(t, b, t.Name())
 
 		b.Publish(t.Name(), []byte("hello"))
 		assert.Equal(t, []byte("hello"), recvWithin(t, ch, recvTimeout))
@@ -44,7 +43,6 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur
 			defer cancel()
 			channels[i] = ch
 		}
-		waitSubscribed(t, b, t.Name()) // all local subscribers share one Redis SUBSCRIBE
 
 		b.Publish(t.Name(), []byte("broadcast"))
 		for i, ch := range channels {
@@ -59,8 +57,6 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur
 		defer cancelA()
 		chB, cancelB := b.Subscribe(topicB)
 		defer cancelB()
-		waitSubscribed(t, b, topicA)
-		waitSubscribed(t, b, topicB)
 
 		b.Publish(topicA, []byte("only-a"))
 		assert.Equal(t, []byte("only-a"), recvWithin(t, chA, recvTimeout))
@@ -100,7 +96,6 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur
 		defer cancelSlow()
 		fast, cancelFast := b.Subscribe(t.Name())
 		defer cancelFast()
-		waitSubscribed(t, b, t.Name())
 
 		// Drain fast concurrently so it keeps up while slow's buffer fills.
 		got := make(chan struct{}, 1)
@@ -147,23 +142,6 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur
 	})
 }
 
-// waitSubscribed blocks until Redis registers the subscription, since
-// RedisBroker.Subscribe returns before the server acks SUBSCRIBE and a publish
-// right after it would race it. Topics are per-scenario, so any subscriber is
-// this one. No-op for MemoryBroker, which registers synchronously.
-func waitSubscribed(t *testing.T, b Broker, topic string) {
-	t.Helper()
-	rb, ok := b.(*RedisBroker)
-	if !ok {
-		return
-	}
-	channel := redisChannelForTopic(topic)
-	require.Eventually(t, func() bool {
-		res, err := rb.client.PubSubNumSub(t.Context(), channel).Result()
-		return err == nil && res[channel] > 0
-	}, 5*time.Second, 10*time.Millisecond, "redis did not register a subscriber on %q", topic)
-}
-
 // recvWithin returns the next message or fails if none arrives before timeout.
 func recvWithin(t *testing.T, ch <-chan []byte, timeout time.Duration) []byte {
 	t.Helper()

services/pubsub/redis.go

@@ -17,10 +17,11 @@ import (
 )
 
 const (
-	redisPingTimeout    = 3 * time.Second
-	redisPingRetries    = 10
-	redisPingRetryDelay = time.Second
-	redisPublishTimeout = 2 * time.Second
+	redisPingTimeout      = 3 * time.Second
+	redisPingRetries      = 10
+	redisPingRetryDelay   = time.Second
+	redisPublishTimeout   = 2 * time.Second
+	redisSubscribeTimeout = 2 * time.Second
 )
 
 // RedisBroker fans out across processes via Redis pub/sub. Each topic is
@@ -94,12 +95,15 @@ func (b *RedisBroker) Subscribe(topic string) (<-chan []byte, func()) {
 	// other Subscribe/cancel calls aren't blocked on the network round-trip.
 	// graceful.ShutdownContext so the reader loop dies cleanly on Gitea
 	// shutdown even if every local subscriber has already cancelled.
-	// readLoop consumes the SUBSCRIBE ack; don't wait for it here, that would put
-	// a Redis round-trip in the WebSocket handshake to close a sub-millisecond
-	// window in which a publish is missed - harmless, since a client receives
-	// nothing at all until its handshake completes.
 	ctx, cancelCtx := context.WithCancel(graceful.GetManager().ShutdownContext())
 	ps := b.client.Subscribe(ctx, redisChannelForTopic(topic))
+	if _, err := ps.ReceiveTimeout(ctx, redisSubscribeTimeout); err != nil { // the ack, so state read afterwards misses no publish
+		log.Error("pubsub redis: subscribe to %q: %v", topic, err)
+		cancelCtx()
+		_ = ps.Close()
+		sub.close()
+		return sub.ch, func() {}
+	}
 	b.mu.Lock()
 	if existing, exists := b.topics[topic]; exists {
 		// Another goroutine won the create race; merge into theirs and discard ours.

services/pubsub/redis_test.go

@@ -59,7 +59,6 @@ func TestRedisBroker(t *testing.T) {
 
 		ch, cancel := subscriber.Subscribe(t.Name())
 		defer cancel()
-		waitSubscribed(t, subscriber, t.Name())
 
 		publisher.Publish(t.Name(), []byte("cross-process"))
 

services/uinotification/notify.go

@@ -55,7 +55,10 @@ func handler(items ...issueNotificationOpts) []issueNotificationOpts {
 	for _, opts := range items {
 		notifiedIDs, err := activities_model.CreateOrUpdateIssueNotifications(ctx, opts.IssueID, opts.CommentID, opts.NotificationAuthorID, opts.ReceiverID)
 		if err != nil {
-			log.Error("Was unable to create issue notification: %v", err)
+			// the issue can be gone by now, e.g. its repository was deleted
+			if !issues_model.IsErrIssueNotExist(err) {
+				log.Error("Unable to create issue notification: %v", err)
+			}
 			continue
 		}
 		for _, userID := range notifiedIDs {

services/websocket/events.go

@@ -1,53 +0,0 @@
-// Copyright 2026 The Gitea Authors. All rights reserved.
-// SPDX-License-Identifier: MIT
-
-package websocket
-
-import (
-	"bytes"
-
-	"gitea.dev/modules/json"
-	"gitea.dev/modules/setting"
-	"gitea.dev/services/pubsub"
-)
-
-// Wire contract with web_src/js/user-events.sharedworker.ts — keep in sync.
-const (
-	EventNotificationCount = "notification-count"
-	EventStopwatches       = "stopwatches"
-	EventLogout            = "logout"
-)
-
-type UserEventMessage[T any] struct {
-	EventType string `json:"eventType"`
-	EventData T      `json:"eventData"`
-}
-
-func publishUserEvent(userID int64, eventType string, eventData any) {
-	if pubsub.DefaultBroker == nil {
-		return
-	}
-	b := MakeUserEventMessage(eventType, eventData)
-	if b == nil {
-		return
-	}
-	pubsub.DefaultBroker.Publish(pubsub.UserTopic(userID), b)
-}
-
-func MakeUserEventMessage(eventType string, eventData any) []byte {
-	buf := &bytes.Buffer{}
-	buf.WriteString(eventType)
-	buf.WriteByte('\n')
-	err := json.MarshalWriteRelaxed(buf, &UserEventMessage[any]{EventType: eventType, EventData: eventData})
-	payloadBytes := bytes.TrimSuffix(buf.Bytes(), []byte("\n")) // json v1 adds extra "\n" but we don't want it
-	if err != nil {
-		setting.PanicInDevOrTesting("websocket: marshal event: %v", err)
-		return nil
-	}
-	return payloadBytes
-}
-
-func ExtractUserEventMessage(b []byte) (string, []byte) {
-	eventType, eventDataBytes, _ := bytes.Cut(b, []byte("\n"))
-	return string(eventType), eventDataBytes
-}

services/websocket/logout.go

@@ -0,0 +1,10 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package websocket
+
+const eventLogout = "logout" // keep in sync with web_src/js/types.ts
+
+func PublishLogout(userID int64, sessionID string) {
+	publishUserEvent(userID, func() []byte { return makeEventMessage(eventLogout, sessionID, nil) })
+}

services/websocket/logout_publisher.go

@@ -1,12 +0,0 @@
-// Copyright 2026 The Gitea Authors. All rights reserved.
-// SPDX-License-Identifier: MIT
-
-package websocket
-
-type LogoutEventData struct {
-	SessionID string `json:"sessionID,omitempty"`
-}
-
-func PublishLogout(userID int64, sessionID string) {
-	publishUserEvent(userID, EventLogout, LogoutEventData{SessionID: sessionID})
-}

services/websocket/notification.go

@@ -9,31 +9,24 @@ import (
 	activities_model "gitea.dev/models/activities"
 	"gitea.dev/models/db"
 	"gitea.dev/modules/log"
-	notify_service "gitea.dev/services/notify"
-	"gitea.dev/services/pubsub"
 )
 
-type notificationCountEventData struct {
-	Count int64 `json:"count"`
-}
-
-type wsNotifier struct {
-	notify_service.NullNotifier
-}
-
-var _ notify_service.Notifier = &wsNotifier{}
+const eventNotificationCount = "notification-count" // keep in sync with web_src/js/types.ts
 
 func (n *wsNotifier) NotificationCountChange(ctx context.Context, userID int64) {
-	if !pubsub.DefaultBroker.HasTopicSubscribers(pubsub.UserTopic(userID)) {
-		return
-	}
+	publishUserEvent(userID, func() []byte { return notificationCountEvent(ctx, userID) })
+}
+
+func notificationCountEvent(ctx context.Context, userID int64) []byte {
 	count, err := db.Count[activities_model.Notification](ctx, activities_model.FindNotificationOptions{
 		UserID: userID,
 		Status: []activities_model.NotificationStatus{activities_model.NotificationStatusUnread},
 	})
 	if err != nil {
 		log.Error("websocket: count notifications for user %d: %v", userID, err)
-		return
+		return nil
 	}
-	publishUserEvent(userID, EventNotificationCount, notificationCountEventData{Count: count})
+	return makeEventMessage(eventNotificationCount, "", struct {
+		Count int64 `json:"count"`
+	}{count})
 }

services/websocket/notifier.go

@@ -1,18 +0,0 @@
-// Copyright 2026 The Gitea Authors. All rights reserved.
-// SPDX-License-Identifier: MIT
-
-package websocket
-
-import (
-	notify_service "gitea.dev/services/notify"
-	"gitea.dev/services/pubsub"
-)
-
-func Init() error {
-	// the pubsub broker must be ready before the notifier starts publishing to it
-	if err := pubsub.Init(); err != nil {
-		return err
-	}
-	notify_service.RegisterNotifier(&wsNotifier{})
-	return nil
-}

services/websocket/stopwatch.go

@@ -10,28 +10,32 @@ import (
 	issues_model "gitea.dev/models/issues"
 	user_model "gitea.dev/models/user"
 	"gitea.dev/modules/log"
+	"gitea.dev/modules/setting"
 	"gitea.dev/modules/util"
 	"gitea.dev/services/convert"
-	"gitea.dev/services/pubsub"
 )
 
-func (n *wsNotifier) StopwatchChanged(ctx context.Context, user *user_model.User) {
-	if !pubsub.DefaultBroker.HasTopicSubscribers(pubsub.UserTopic(user.ID)) {
-		return
-	}
+const eventStopwatches = "stopwatches" // keep in sync with web_src/js/types.ts
 
+func (n *wsNotifier) StopwatchChanged(ctx context.Context, user *user_model.User) {
+	publishUserEvent(user.ID, func() []byte { return stopwatchesEvent(ctx, user) })
+}
+
+func stopwatchesEvent(ctx context.Context, user *user_model.User) []byte {
+	if !setting.Service.EnableTimetracking {
+		return nil
+	}
 	sws, err := issues_model.GetUserStopwatches(ctx, user.ID, db.ListOptions{})
 	if err != nil {
 		log.Error("websocket: GetUserStopwatches %d: %v", user.ID, err)
-		return
+		return nil
 	}
-
 	apiStopWatches, err := convert.ToStopWatches(ctx, user, sws)
 	if err != nil {
 		if !issues_model.IsErrIssueNotExist(err) {
 			log.Error("websocket: ToStopWatches: %v", err)
 		}
-		return
+		return nil
 	}
-	publishUserEvent(user.ID, EventStopwatches, util.SliceNilAsEmpty(apiStopWatches))
+	return makeEventMessage(eventStopwatches, "", util.SliceNilAsEmpty(apiStopWatches))
 }

services/websocket/websocket.go

@@ -0,0 +1,71 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package websocket
+
+import (
+	"bytes"
+	"context"
+
+	user_model "gitea.dev/models/user"
+	"gitea.dev/modules/json"
+	"gitea.dev/modules/setting"
+	notify_service "gitea.dev/services/notify"
+	"gitea.dev/services/pubsub"
+)
+
+type wsNotifier struct {
+	notify_service.NullNotifier
+}
+
+var _ notify_service.Notifier = &wsNotifier{}
+
+func Init() error {
+	// the pubsub broker must be ready before the notifier starts publishing to it
+	if err := pubsub.Init(); err != nil {
+		return err
+	}
+	notify_service.RegisterNotifier(&wsNotifier{})
+	return nil
+}
+
+func UserData(ctx context.Context, user *user_model.User) (data [][]byte) {
+	for _, event := range [][]byte{notificationCountEvent(ctx, user.ID), stopwatchesEvent(ctx, user)} {
+		if event != nil {
+			data = append(data, event)
+		}
+	}
+	return data
+}
+
+func EventForSession(brokerPayload []byte, sessionID string) []byte {
+	target, eventData, _ := bytes.Cut(brokerPayload, []byte("\n"))
+	if len(target) > 0 && string(target) != sessionID {
+		return nil
+	}
+	return eventData
+}
+
+// makeEventMessage prefixes the client JSON with its target session, empty for all sessions.
+func makeEventMessage(eventType, targetSession string, eventData any) []byte {
+	buf := bytes.NewBufferString(targetSession + "\n")
+	err := json.MarshalWriteRelaxed(buf, &struct {
+		EventType string `json:"eventType"`
+		EventData any    `json:"eventData,omitzero"`
+	}{eventType, eventData})
+	if err != nil {
+		setting.PanicInDevOrTesting("websocket: marshal event: %v", err)
+		return nil
+	}
+	return buf.Bytes()
+}
+
+func publishUserEvent(userID int64, build func() []byte) {
+	topic := pubsub.UserTopic(userID)
+	if !pubsub.DefaultBroker.HasTopicSubscribers(topic) {
+		return
+	}
+	if event := build(); event != nil {
+		pubsub.DefaultBroker.Publish(topic, event)
+	}
+}

services/websocket/websocket_test.go

@@ -0,0 +1,19 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package websocket
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestEventForSessionTargetsAndKeepsEmptyData(t *testing.T) {
+	targeted := makeEventMessage("targeted-event", "sess-A", nil)
+	assert.JSONEq(t, `{"eventType":"targeted-event"}`, string(EventForSession(targeted, "sess-A")))
+	assert.Nil(t, EventForSession(targeted, "sess-B"))
+
+	broadcast := makeEventMessage("list-event", "", []int{})
+	assert.JSONEq(t, `{"eventType":"list-event","eventData":[]}`, string(EventForSession(broadcast, "sess-A")))
+}

tests/e2e/events.test.ts

@@ -23,7 +23,7 @@ test.describe('events', () => {
     await expect(page.locator('html[data-user-events-connected]')).toBeAttached();
 
     await apiCreateIssue(request, {owner, repo: repoName, title: 'events-notif', headers: apiUserHeaders(commenter)});
-    await expect(badge).toBeVisible();
+    await expect(badge).toHaveText('1');
   });
 
   test('stopwatch appears via real-time push and stops from its popup', async ({page, request}) => {
@@ -51,6 +51,7 @@ test.describe('events', () => {
     await expect(stopwatch).toBeVisible();
 
     await stopwatch.click();
+    await expect(page.locator('.stopwatch-link')).toHaveAttribute('href', `/${name}/${name}/issues/1`);
     await page.getByRole('button', {name: 'Stop Timer'}).click();
     await expect(stopwatch).toBeHidden();
   });

tests/integration/websocket_test.go

@@ -0,0 +1,79 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package integration
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	activities_model "gitea.dev/models/activities"
+	"gitea.dev/models/unittest"
+	"gitea.dev/modules/json"
+	"gitea.dev/modules/setting"
+	"gitea.dev/modules/test"
+	"gitea.dev/services/pubsub"
+	"gitea.dev/tests"
+
+	"github.com/coder/websocket"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type staleBacklogBroker struct {
+	pubsub.Broker
+}
+
+func (b *staleBacklogBroker) Subscribe(topic string) (<-chan []byte, func()) {
+	ch, cancel := b.Broker.Subscribe(topic)
+	for range 9 {
+		b.Publish(topic, countEvent(-1))
+	}
+	return ch, cancel
+}
+
+func countEvent(count int) []byte {
+	return fmt.Appendf(nil, "\n"+`{"eventType":"notification-count","eventData":{"count":%d}}`, count)
+}
+
+func TestWebSocketSnapshotFollowsBacklog(t *testing.T) {
+	defer tests.PrepareTestEnv(t)()
+	defer test.MockVariableValue(&pubsub.DefaultBroker, pubsub.Broker(&staleBacklogBroker{pubsub.NewMemoryBroker()}))()
+	session := loginUser(t, "user2")
+	server := httptest.NewServer(testWebRoutes)
+	defer server.Close()
+
+	header := http.Header{}
+	header.Set("Cookie", session.GetRawCookie(setting.SessionConfig.CookieName).String())
+	conn, _, err := websocket.Dial(t.Context(), strings.Replace(server.URL, "http", "ws", 1)+"/-/ws", &websocket.DialOptions{HTTPHeader: header}) //nolint:bodyclose // coder/websocket owns the body, it is nil after the upgrade
+	require.NoError(t, err)
+	defer conn.CloseNow()
+
+	const sentinelCount = -2
+	lastCount := 0
+	for {
+		_, payload, err := conn.Read(t.Context())
+		require.NoError(t, err)
+		var event struct {
+			EventType string     `json:"eventType"`
+			EventData json.Value `json:"eventData"`
+		}
+		require.NoError(t, json.Unmarshal(payload, &event))
+		if event.EventType == "stopwatches" {
+			pubsub.DefaultBroker.Publish(pubsub.UserTopic(2), countEvent(sentinelCount))
+			continue
+		}
+		var data struct {
+			Count int `json:"count"`
+		}
+		require.NoError(t, json.Unmarshal(event.EventData, &data))
+		if data.Count == sentinelCount {
+			break
+		}
+		lastCount = data.Count
+	}
+	assert.Equal(t, unittest.GetCount(t, &activities_model.Notification{UserID: 2, Status: activities_model.NotificationStatusUnread}), lastCount)
+}

web_src/js/features/notification.ts

@@ -26,8 +26,7 @@ export function initNotificationCount() {
 
   let pollerStarted = false;
   onUserEvent('notification-count', (msg) => { receiveUpdateCount(msg.eventData.count) }); // no await
-  // On each (re)connect, reconcile the count from the server to recover any push dropped during the connect gap.
-  onUserEvent('worker-connected', () => { updateNotificationCount(); updateNotificationTable() }); // no await
+  onUserEvent('worker-attached', () => { updateNotificationCount(); updateNotificationTable() }); // no await
   onUserEvent('worker-unavailable', () => {
     if (pollerStarted) return;
     pollerStarted = true;

web_src/js/features/stopwatch.ts

@@ -76,8 +76,7 @@ export const initStopwatch = () => registerGlobalInitFunc('initActiveStopwatchNo
 
   let pollerStarted = false;
   onUserEvent('stopwatches', (msg) => updateStopwatchData(msg.eventData));
-  // On each (re)connect, reconcile stopwatch state from the server to recover any push dropped during the connect gap.
-  onUserEvent('worker-connected', () => { updateStopwatch() }); // no await
+  onUserEvent('worker-attached', () => { updateStopwatch() }); // no await
   onUserEvent('worker-unavailable', () => {
     if (pollerStarted) return;
     pollerStarted = true;

web_src/js/types.ts

@@ -87,7 +87,7 @@ export type StopwatchData = {
   seconds: number,
 };
 
-// keep in sync with services/websocket/events.go
+// keep in sync with services/websocket
 export type ServerUserEventMessage =
   {eventType: 'notification-count', eventData: {count: number}} |
   {eventType: 'stopwatches', eventData: Array<StopwatchData>} |
@@ -97,7 +97,8 @@ export const serverUserEventTypes = ['notification-count', 'stopwatches', 'logou
 
 export type UserEventMessage = ServerUserEventMessage |
   {eventType: 'worker-unavailable'} |
-  {eventType: 'worker-connected'};
+  {eventType: 'worker-connected'} |
+  {eventType: 'worker-attached'}; // joined an open socket after its connect snapshot
 
 export type UserEventType = UserEventMessage['eventType'];
 

web_src/js/user-events.sharedworker.ts

@@ -96,9 +96,6 @@ class WsSource {
     this.ws.addEventListener('open', () => {
       this.reconnectDelay = 1000;
       this.failuresWithoutConnect = 0;
-      // Pushes fired while no client was subscribed (initial connect gap, or a
-      // reconnect window) are dropped server-side, so tell clients to reconcile
-      // their state from the server on every fresh connection.
       this.source.notifyClientsUserEvent({eventType: 'worker-connected'});
     });
 
@@ -191,6 +188,7 @@ const wsSourcesByUrl = new Map<string, WsSource>();
           const openWs = wsSourcesByUrl.get(url);
           if (openWs?.ws?.readyState === WebSocket.OPEN) {
             postUserEventMessage(port, {eventType: 'worker-connected'});
+            postUserEventMessage(port, {eventType: 'worker-attached'});
           }
           return;
         }