Skip to content

Commit 50c9b19

Browse files
committed
subscribe pingpong
1 parent 5652d23 commit 50c9b19

2 files changed

Lines changed: 21 additions & 6 deletions

File tree

client/remote/client.go

Lines changed: 20 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1455,14 +1455,29 @@ func (c *Client) writeLoop() {
14551455
err := conn.WriteMessage(websocket.BinaryMessage, data)
14561456
if err != nil {
14571457
c.logger.Error("WebSocket write error", zap.Error(err))
1458+
1459+
// If the error indicates we already sent a close, treat this as a normal
1460+
// closure for PUSH syncs and mark the sync completed so the caller can
1461+
// finish processing. This happens when the websocket library returns
1462+
// "websocket: close sent" while attempting to write after a close.
1463+
if strings.Contains(err.Error(), "close sent") {
1464+
c.logger.Info("Write error contains 'close sent' - treating as normal closure")
1465+
c.logger.Info("Ending PUSH sync due to close-sent write error")
1466+
c.setSyncCompleted(true)
1467+
}
14581468
c.setError(err)
14591469
c.setConnected(false)
14601470

1461-
// Signal potential reconnection
1462-
select {
1463-
case c.reconnectChan <- struct{}{}:
1464-
default:
1465-
}
1471+
// Treat this as a true disconnection (do not trigger reconnect).
1472+
// Close the read queue so any readers observe EOF and stop.
1473+
func() {
1474+
defer func() {
1475+
if r := recover(); r != nil {
1476+
// ignore if already closed
1477+
}
1478+
}()
1479+
close(c.readQueue)
1480+
}()
14661481
return
14671482
}
14681483

client/sync/coordinator.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -361,7 +361,7 @@ func (c *Coordinator) executePull(isSubscription bool) error {
361361
AuthKey: authResult.AccessKey,
362362
ReplicaID: authResult.ReplicaID,
363363
Timeout: 8000,
364-
PingPong: false, // No ping/pong needed for single sync
364+
PingPong: true, // Ping/pong enabled for subscription sync
365365
Logger: c.logger.Named("remote"),
366366
Subscribe: false, // Subscription handled separately
367367
EnableTrafficInspection: c.config.Verbose,

0 commit comments

Comments
 (0)