Potential Bug: Offset Not Updated for Server-Side Subscriptions After Receiving Publications

[shalom-aviv]

Description:

We have encountered an issue where, after manually reconnecting the client, the same chunk of history is repeatedly retrieved for a server-side subscription.

In debugging, we noticed that the offset sent during the connection request for a channel is the same as the offset returned in the server response, even when publications for that channel are delivered during the connection. This behavior indicates that while the messages are successfully delivered to the client, the offset for the channel is not updated accordingly.

        if let result = res {
            self.state = .connected
            self.reconnectAttempts = 0
            self.client = result.client
            self.delegate?.onConnected(self, CentrifugeConnectedEvent(client: result.client, data: result.data))
            for cb in self.connectCallbacks.values {
                cb(nil)
            }
            self.connectCallbacks.removeAll(keepingCapacity: false)
            
            // Process server-side subscriptions.
            for (channel, subResult) in result.subs {
/// NOTE:
/// Save all parameters for the server-side channel received from the server.
/// In debugging, we observe that the offset here is the same as the one sent during the connection,
/// meaning it does not change.
/// Therefore, if no publications are received from the server for this channel after connection,
/// on the next reconnection, the same chunk of history will be retrieved because the offset remains unchanged.
                self.serverSubs[channel] = ServerSubscription(recoverable: subResult.recoverable, offset: subResult.offset, epoch: subResult.epoch)
                let event = CentrifugeServerSubscribedEvent(channel: channel, wasRecovering: subResult.wasRecovering, recovered: subResult.recovered, positioned: subResult.positioned, recoverable: subResult.recoverable, streamPosition: subResult.positioned || subResult.recoverable ? StreamPosition(offset: subResult.offset, epoch: subResult.epoch) : nil, data: subResult.data)
                self.delegate?.onSubscribed(self, event)
                subResult.publications.forEach{ pub in
                    var info: CentrifugeClientInfo? = nil;
                    if pub.hasInfo {
                        info = CentrifugeClientInfo(client: pub.info.client, user: pub.info.user, connInfo: pub.info.connInfo, chanInfo: pub.info.chanInfo)
                    }
                    let pubEvent = CentrifugeServerPublicationEvent(channel: channel, data: pub.data, offset: pub.offset, tags: pub.tags, info: info)
                    self.delegate?.onPublication(self, pubEvent)
/// NOTE:
/// We notified about receiving the publication via .onPublication
/// but did not update the offset for the channel  like in:
//// private func handlePub(channel: String, pub: Centrifugal_Centrifuge_Protocol_Publication) {
/// ...
///                if self.serverSubs[channel]?.recoverable == true && pub.offset > 0 {
///                    self.serverSubs[channel]?.offset = pub.offset
///                }
///  ....
/// }
                }
//.....
            }
//.....
        }

[FZambia]

Hello @shalom-aviv

Yes, it's a bug. Thanks! Please send PR if you have time, or I'll fix it very soon.

[hikkidev]

This issue also exists in https://github.com/centrifugal/centrifuge-java

[shalom-aviv]

Hello @shalom-aviv

Yes, it's a bug. Thanks! Please send PR if you have time, or I'll fix it very soon.

#108

Advice from Spikeman:

Before releasing the fix on the client, as a temporary workaround, the offset for publications in the target server channel can be stored and previously delivered messages can be skipped.

[FZambia]

This issue also exists in https://github.com/centrifugal/centrifuge-java

Hello @hikkidev - I think it does not - https://github.com/centrifugal/centrifuge-java/blob/d0678bff8740a79eb84d2ca74973d4dc9e285f79/centrifuge/src/main/java/io/github/centrifugal/centrifuge/Client.java#L696 - could you elaborate?

[FZambia]

The fix is part of https://github.com/centrifugal/centrifuge-swift/releases/tag/0.7.3

Many thanks!