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

Backport of https://github.com/go-gitea/gitea/pull/38847, which fixes
the stopwatch push flake in
https://github.com/go-gitea/gitea/actions/runs/43200377380/job/131122930118.

The stopwatch e2e test keeps its v28 flow, since main reworked it in a
separate change.

Assisted-by: Claude Code:claude-opus-5-5

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 and hides via real-time push', async ({page, request}) => {

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

@@ -48,8 +48,7 @@ export function initStopwatch() {
 
   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;
         }