diff --git a/server/channels/app/platform/web_hub.go b/server/channels/app/platform/web_hub.go index 948272f42f..5eab747a1c 100644 --- a/server/channels/app/platform/web_hub.go +++ b/server/channels/app/platform/web_hub.go @@ -311,7 +311,7 @@ func (ps *PlatformService) CheckWebConn(userID, connectionID string, seqNum int6 connRes.ActiveQueue = aq connRes.ReuseCount = queues.ReuseCount - // parse the dq, wc.addToDeadQ() + // parse the deadq if queues.DeadQ != nil { dq, dqPtr, err := ps.UnmarshalDQ(queues.DeadQ) if err != nil { @@ -322,7 +322,9 @@ func (ps *PlatformService) CheckWebConn(userID, connectionID string, seqNum int6 return nil } - if dqPtr > 0 { + // We check if atleast one item has been written. + // Length of dq is always guaranteed to be deadQueueSize. + if dq[0] != nil { connRes.DeadQueue = dq connRes.DeadQueuePointer = dqPtr } diff --git a/server/channels/app/platform/websocket_reliable.go b/server/channels/app/platform/websocket_reliable.go index 09a269d85c..0fe8653137 100644 --- a/server/channels/app/platform/websocket_reliable.go +++ b/server/channels/app/platform/websocket_reliable.go @@ -156,7 +156,7 @@ func (ps *PlatformService) UnmarshalDQ(buf []json.RawMessage) ([]*model.WebSocke // Same as active queue, this can never be out of bounds because all dead queues // are of deadQueueSize. dq[dqPtr] = item - dqPtr++ + dqPtr = (dqPtr + 1) % deadQueueSize } return dq, dqPtr, nil } diff --git a/server/channels/app/platform/websocket_reliable_test.go b/server/channels/app/platform/websocket_reliable_test.go index c9cbe333b9..e9ced119e5 100644 --- a/server/channels/app/platform/websocket_reliable_test.go +++ b/server/channels/app/platform/websocket_reliable_test.go @@ -65,3 +65,54 @@ func TestMarshalDQ(t *testing.T) { assert.Equal(t, 3, dqPtr) assert.Equal(t, events[:3], gotEvents[:3]) } + +func TestUnmarshalDQFullBuffer(t *testing.T) { + ps := PlatformService{} + + t.Run("dq full", func(t *testing.T) { + // Create exactly deadQueueSize events + events := make([]*model.WebSocketEvent, deadQueueSize) + for i := 0; i < deadQueueSize; i++ { + events[i] = model.NewWebSocketEvent(model.WebsocketEventPosted, "t1", "c1", "u1", nil, "").SetSequence(int64(i)) + } + + // Set up a scenario where the buffer is already filled and has wrapped around + // Use index 0 and simulate that the dqPtr has wrapped around to 0 again + got, err := ps.marshalDQ(events, 0, 0) + require.NoError(t, err) + require.Len(t, got, deadQueueSize) + + // Unmarshal the full buffer back + gotEvents, dqPtr, err := ps.UnmarshalDQ(got) + require.NoError(t, err) + + // Check that dqPtr wraps around to 0, not deadQueueSize + assert.Equal(t, 0, dqPtr, "dqPtr should be 0 for a full buffer (deadQueueSize % deadQueueSize = 0)") + + // Verify all events were unmarshaled correctly + assert.Equal(t, events, gotEvents) + }) + + t.Run("dq rollover", func(t *testing.T) { + // Alternative test: Create a simulation of the circular buffer behavior + // This test fills up to the max and ensures wraparound works correctly + events := make([]*model.WebSocketEvent, deadQueueSize) + for i := 0; i < deadQueueSize; i++ { + // Create events with sequence numbers that show wraparound + // Last event will have highest sequence to demonstrate the break condition + // Seq nos: 100 - 228 + events[i] = model.NewWebSocketEvent(model.WebsocketEventPosted, "t1", "c1", "u1", nil, "").SetSequence(int64(i + 100)) + } + + // Marshal only the last entry wrapping to the first to test wraparound detection + got2, err := ps.marshalDQ(events, deadQueueSize-1, 0) + require.NoError(t, err) + require.Len(t, got2, 1) // Just the last element + + // Unmarshal this single element + gotEvents2, dqPtr2, err := ps.UnmarshalDQ(got2) + require.NoError(t, err) + assert.Equal(t, 1, dqPtr2, "dqPtr should be 1 for a 1-element buffer (1 % deadQueueSize = 1)") + assert.Equal(t, events[deadQueueSize-1], gotEvents2[0]) + }) +}