From 7308c5fb974cdfa48b2c656f3e1da7761f69d71a Mon Sep 17 00:00:00 2001 From: silverwind Date: Thu, 29 Oct 2026 06:18:17 +0100 Subject: [PATCH] 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 | 80 ++++++++++--------- routers/web/websocket/websocket_test.go | 60 -------------- services/pubsub/broker_test.go | 22 ----- services/pubsub/redis.go | 20 +++-- services/pubsub/redis_test.go | 1 - services/uinotification/notify.go | 5 +- services/websocket/events.go | 53 ------------ services/websocket/logout.go | 10 +++ services/websocket/logout_publisher.go | 12 --- ...tification_notifier.go => notification.go} | 25 +++--- services/websocket/notifier.go | 18 ----- .../{stopwatch_notifier.go => stopwatch.go} | 22 ++--- services/websocket/websocket.go | 71 ++++++++++++++++ services/websocket/websocket_test.go | 19 +++++ tests/e2e/events.test.ts | 3 +- tests/integration/websocket_test.go | 79 ++++++++++++++++++ web_src/js/features/notification.ts | 3 +- web_src/js/features/stopwatch.ts | 3 +- web_src/js/types.ts | 5 +- web_src/js/user-events.sharedworker.ts | 4 +- 20 files changed, 268 insertions(+), 247 deletions(-) delete mode 100644 routers/web/websocket/websocket_test.go delete mode 100644 services/websocket/events.go create mode 100644 services/websocket/logout.go delete mode 100644 services/websocket/logout_publisher.go rename services/websocket/{notification_notifier.go => notification.go} (59%) delete mode 100644 services/websocket/notifier.go rename services/websocket/{stopwatch_notifier.go => stopwatch.go} (63%) create mode 100644 services/websocket/websocket.go create mode 100644 services/websocket/websocket_test.go create mode 100644 tests/integration/websocket_test.go diff --git a/routers/web/websocket/websocket.go b/routers/web/websocket/websocket.go index cc9c75bc3c3..553ffa7191d 100644 --- a/routers/web/websocket/websocket.go +++ b/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") +} diff --git a/routers/web/websocket/websocket_test.go b/routers/web/websocket/websocket_test.go deleted file mode 100644 index 421fb2f7633..00000000000 --- a/routers/web/websocket/websocket_test.go +++ /dev/null @@ -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) - }) - } -} diff --git a/services/pubsub/broker_test.go b/services/pubsub/broker_test.go index d2fc36e9dc1..b09d70ed628 100644 --- a/services/pubsub/broker_test.go +++ b/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() diff --git a/services/pubsub/redis.go b/services/pubsub/redis.go index eb7798298e2..f1b90368ddd 100644 --- a/services/pubsub/redis.go +++ b/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. diff --git a/services/pubsub/redis_test.go b/services/pubsub/redis_test.go index f1ad3a7512e..7b9855c5995 100644 --- a/services/pubsub/redis_test.go +++ b/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")) diff --git a/services/uinotification/notify.go b/services/uinotification/notify.go index b0aa8c1ac32..1823da09e3e 100644 --- a/services/uinotification/notify.go +++ b/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 { diff --git a/services/websocket/events.go b/services/websocket/events.go deleted file mode 100644 index c7c3ec052eb..00000000000 --- a/services/websocket/events.go +++ /dev/null @@ -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 -} diff --git a/services/websocket/logout.go b/services/websocket/logout.go new file mode 100644 index 00000000000..90e65d9de47 --- /dev/null +++ b/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) }) +} diff --git a/services/websocket/logout_publisher.go b/services/websocket/logout_publisher.go deleted file mode 100644 index 58c0bb3962a..00000000000 --- a/services/websocket/logout_publisher.go +++ /dev/null @@ -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}) -} diff --git a/services/websocket/notification_notifier.go b/services/websocket/notification.go similarity index 59% rename from services/websocket/notification_notifier.go rename to services/websocket/notification.go index 7013dc2b637..f82ce2be5e8 100644 --- a/services/websocket/notification_notifier.go +++ b/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}) } diff --git a/services/websocket/notifier.go b/services/websocket/notifier.go deleted file mode 100644 index ec09c4f0d2f..00000000000 --- a/services/websocket/notifier.go +++ /dev/null @@ -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 -} diff --git a/services/websocket/stopwatch_notifier.go b/services/websocket/stopwatch.go similarity index 63% rename from services/websocket/stopwatch_notifier.go rename to services/websocket/stopwatch.go index 5d81cbd139e..b9f8d83ad42 100644 --- a/services/websocket/stopwatch_notifier.go +++ b/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)) } diff --git a/services/websocket/websocket.go b/services/websocket/websocket.go new file mode 100644 index 00000000000..43048bf7731 --- /dev/null +++ b/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) + } +} diff --git a/services/websocket/websocket_test.go b/services/websocket/websocket_test.go new file mode 100644 index 00000000000..5d7d367a637 --- /dev/null +++ b/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"))) +} diff --git a/tests/e2e/events.test.ts b/tests/e2e/events.test.ts index d48bcccc3bc..026e489ebfd 100644 --- a/tests/e2e/events.test.ts +++ b/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(); }); diff --git a/tests/integration/websocket_test.go b/tests/integration/websocket_test.go new file mode 100644 index 00000000000..fed6db20605 --- /dev/null +++ b/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) +} diff --git a/web_src/js/features/notification.ts b/web_src/js/features/notification.ts index f18567e62ba..274bc66e670 100644 --- a/web_src/js/features/notification.ts +++ b/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; diff --git a/web_src/js/features/stopwatch.ts b/web_src/js/features/stopwatch.ts index 2651856420d..43498a12210 100644 --- a/web_src/js/features/stopwatch.ts +++ b/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; diff --git a/web_src/js/types.ts b/web_src/js/types.ts index 5793cfd2372..833ef089aaf 100644 --- a/web_src/js/types.ts +++ b/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} | @@ -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']; diff --git a/web_src/js/user-events.sharedworker.ts b/web_src/js/user-events.sharedworker.ts index 2a4ca963c32..bf2540a0eaf 100644 --- a/web_src/js/user-events.sharedworker.ts +++ b/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(); const openWs = wsSourcesByUrl.get(url); if (openWs?.ws?.readyState === WebSocket.OPEN) { postUserEventMessage(port, {eventType: 'worker-connected'}); + postUserEventMessage(port, {eventType: 'worker-attached'}); } return; }