From 2bc4ce7060d2f6e3537bd24dd80c37f4fd4461ec Mon Sep 17 00:00:00 2001 From: Arjun Patel Date: Sun, 12 Apr 2026 10:00:11 -0700 Subject: [PATCH] fix: unreliable websocket functionality (#144) * increases timeout * refactor: reuse constants for redis db * avoid closing connection on pong error * fix: prevent electron from pausing timer executions Our pings could be paused when our app is backgrounded. * refactor: remove unnecessary refs * fix: presence unreliable within same pod It was broadcasting presence changes to other pods, but not notifying clients connected to the same pod --- go/cmd/orion/main.go | 7 +---- go/cmd/pusherservice/main.go | 19 +++++-------- go/internal/db/{connect.go => postgres.go} | 0 go/internal/db/redis.go | 8 ++++++ go/internal/pusher/conn.go | 5 ++-- go/internal/pusher/hub.go | 31 ++++++++++++++++++++++ go/internal/pusher/server.go | 12 ++++++--- go/k8s/dev/pusher.yaml | 2 +- go/k8s/prod/pusher.yaml | 2 +- js/src/hooks/use-channel.ts | 12 +++------ js/src/main.ts | 1 + 11 files changed, 64 insertions(+), 35 deletions(-) rename go/internal/db/{connect.go => postgres.go} (100%) create mode 100644 go/internal/db/redis.go diff --git a/go/cmd/orion/main.go b/go/cmd/orion/main.go index 6cd2183..eb1fc22 100644 --- a/go/cmd/orion/main.go +++ b/go/cmd/orion/main.go @@ -27,13 +27,8 @@ import ( "google.golang.org/grpc/credentials/insecure" ) -const ( - // FIX: Use separate redis instance. We start with higher number because use this same instance in helios. - REDIS_DATABASE_FOR_AUTH int = 4 -) - func redisForAuth() *redis.Client { - return internal.ConnectAndTestRedis(REDIS_DATABASE_FOR_AUTH) + return internal.ConnectAndTestRedis(db.RedisDBAuth) } func main() { diff --git a/go/cmd/pusherservice/main.go b/go/cmd/pusherservice/main.go index 816bb9d..2d0c253 100644 --- a/go/cmd/pusherservice/main.go +++ b/go/cmd/pusherservice/main.go @@ -21,11 +21,6 @@ import ( "google.golang.org/grpc" ) -const ( - redisDBAuth = 4 // shared with orion for session validation - redisDBPusher = 5 // dedicated to pusher state (presence, pub/sub) -) - func main() { port := utils.MustGetEnv("PORT") grpcPort := utils.MustGetEnv("GRPC_PORT") @@ -35,10 +30,10 @@ func main() { defer db.Cleanup() // Redis for auth session validation (same DB as orion) - authRedis := internal.ConnectAndTestRedis(redisDBAuth) + authRedis := internal.ConnectAndTestRedis(db.RedisDBAuth) // Redis for pusher state (presence hashes, pub/sub) - pusherRedis := internal.ConnectAndTestRedis(redisDBPusher) + pusherRedis := internal.ConnectAndTestRedis(db.RedisDBPusher) // Services authSvc := auth.NewAuthService(authRedis, nil) // nil aeroSvc — pusher only calls GetSession @@ -52,15 +47,15 @@ func main() { // Pusher core bridge := pusher.NewRedisBridge(pusherRedis, podID) - authorizer := pusher.NewAuthorizer(networkSvc) - hub := pusher.NewHub(bridge, authorizer) - bridge.SetHub(hub) - server := pusher.NewServer(hub, bridge, authSvc) - // Context for graceful shutdown ctx, cancel := context.WithCancel(context.Background()) defer cancel() + authorizer := pusher.NewAuthorizer(networkSvc) + hub := pusher.NewHub(bridge, authorizer) + bridge.SetHub(hub) + server := pusher.NewServer(ctx, hub, bridge, authSvc) + // Start hub event loop go hub.Run(ctx) diff --git a/go/internal/db/connect.go b/go/internal/db/postgres.go similarity index 100% rename from go/internal/db/connect.go rename to go/internal/db/postgres.go diff --git a/go/internal/db/redis.go b/go/internal/db/redis.go new file mode 100644 index 0000000..d663685 --- /dev/null +++ b/go/internal/db/redis.go @@ -0,0 +1,8 @@ +package db + +// Shared database namespaces used across services +// FIX: Use separate redis instance. We start with higher number because use this same instance in helios. +const ( + RedisDBAuth = 4 // auth sessions + RedisDBPusher = 5 // dedicated to pusher state (presence, pub/sub) +) diff --git a/go/internal/pusher/conn.go b/go/internal/pusher/conn.go index 89d46c7..3cdf2c0 100644 --- a/go/internal/pusher/conn.go +++ b/go/internal/pusher/conn.go @@ -41,15 +41,14 @@ func (c *Conn) ReadPump(ctx context.Context, hub *Hub) { slog.Info("websocket context cancelled", "connId", c.id, "humanId", c.humanID, "error", ctx.Err()) return } - slog.Info("websocket read error", "connId", c.id, "humanId", c.humanID, "error", err) + slog.Warn("websocket read error", "connId", c.id, "humanId", c.humanID, "error", err) return } // Respond to keep-alive pings if string(data) == "ping" { if err := c.ws.Write(ctx, websocket.MessageText, []byte("pong")); err != nil { - slog.Debug("websocket pong write error", "connId", c.id, "error", err) - return + slog.Warn("websocket pong write error", "connId", c.id, "error", err) } continue } diff --git a/go/internal/pusher/hub.go b/go/internal/pusher/hub.go index 480e019..1de54e4 100644 --- a/go/internal/pusher/hub.go +++ b/go/internal/pusher/hub.go @@ -101,6 +101,9 @@ func (h *Hub) handleSubscribe(ctx context.Context, req *subscribeRequest) { h.channels[req.channelID] = ch } + // Capture before addMember so multi-tab joins don't emit a spurious join. + wasPresentLocally := ch.hasHumanID(req.conn.humanID) + // Add to local channel ch.addMember(req.conn, req.conn.humanID) @@ -124,6 +127,16 @@ func (h *Hub) handleSubscribe(ctx context.Context, req *subscribeRequest) { Channel: req.channelID, Presence: presence, }) + + // Notify other local members. The Redis self-filter drops our own echo, + // so same-pod peers would otherwise never hear about this join. + if !wasPresentLocally { + ch.broadcast(ServerMessage{ + Type: TypeJoin, + Channel: req.channelID, + HumanID: req.conn.humanID, + }, req.conn) + } } func (h *Hub) handleUnsubscribe(ctx context.Context, req *unsubscribeRequest) { @@ -144,6 +157,16 @@ func (h *Hub) handleUnsubscribe(ctx context.Context, req *unsubscribeRequest) { slog.Error("redis unsubscribe failed", "channelId", req.channelID, "error", err) } + // Notify other local members iff the humanID is fully gone from this pod + // (multi-tab: other conns keep them present, so no leave fires). + if !ch.hasHumanID(req.conn.humanID) { + ch.broadcast(ServerMessage{ + Type: TypeLeave, + Channel: req.channelID, + HumanID: req.conn.humanID, + }, req.conn) + } + // Clean up empty local channel if ch.isEmpty() { delete(h.channels, req.channelID) @@ -192,6 +215,14 @@ func (h *Hub) handleDisconnect(ctx context.Context, conn *Conn) { slog.Error("redis unsubscribe on disconnect failed", "channelId", channelID, "error", err) } + if !ch.hasHumanID(conn.humanID) { + ch.broadcast(ServerMessage{ + Type: TypeLeave, + Channel: channelID, + HumanID: conn.humanID, + }, conn) + } + if ch.isEmpty() { delete(h.channels, channelID) } diff --git a/go/internal/pusher/server.go b/go/internal/pusher/server.go index 37b997f..1b4dd9d 100644 --- a/go/internal/pusher/server.go +++ b/go/internal/pusher/server.go @@ -15,14 +15,17 @@ import ( type Server struct { pbpusher.UnimplementedPusherServiceServer + ctx context.Context // server-scoped context for graceful shutdown hub *Hub bridge *RedisBridge authSvc auth.AuthService } -// NewServer creates a new pusher server. -func NewServer(hub *Hub, bridge *RedisBridge, authSvc auth.AuthService) *Server { +// NewServer creates a new pusher server. The ctx controls the lifetime of all +// WebSocket connections — when cancelled, all connections are closed gracefully. +func NewServer(ctx context.Context, hub *Hub, bridge *RedisBridge, authSvc auth.AuthService) *Server { return &Server{ + ctx: ctx, hub: hub, bridge: bridge, authSvc: authSvc, @@ -59,7 +62,10 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) { slog.Info("websocket connected", "connId", connID, "humanId", session.HumanId) - ctx, cancel := context.WithCancel(r.Context()) + // Use server context, NOT r.Context(). After WebSocket upgrade, the HTTP + // request context can be cancelled by load balancers or Go's HTTP server, + // and nhooyr/websocket permanently closes the conn on any context error. + ctx, cancel := context.WithCancel(s.ctx) defer cancel() // Auto-subscribe to presence channel so this user appears online diff --git a/go/k8s/dev/pusher.yaml b/go/k8s/dev/pusher.yaml index 8548375..3208b18 100644 --- a/go/k8s/dev/pusher.yaml +++ b/go/k8s/dev/pusher.yaml @@ -88,7 +88,7 @@ metadata: name: pusher-backend-policy spec: default: - timeoutSec: 60 + timeoutSec: 3600 connectionDraining: drainingTimeoutSec: 30 targetRef: diff --git a/go/k8s/prod/pusher.yaml b/go/k8s/prod/pusher.yaml index e4649a5..9e5591e 100644 --- a/go/k8s/prod/pusher.yaml +++ b/go/k8s/prod/pusher.yaml @@ -85,7 +85,7 @@ metadata: name: pusher-backend-policy spec: default: - timeoutSec: 60 + timeoutSec: 3600 connectionDraining: drainingTimeoutSec: 30 targetRef: diff --git a/js/src/hooks/use-channel.ts b/js/src/hooks/use-channel.ts index 0cc1fd7..2971ed4 100644 --- a/js/src/hooks/use-channel.ts +++ b/js/src/hooks/use-channel.ts @@ -22,12 +22,6 @@ export function useChannel(channelId: string | null): UseChannelResult { const [presence, setPresence] = useState([]); const [messages, setMessages] = useState([]); - // Keep a ref to avoid re-subscribing when sendMessage changes - const clientRef = useRef(client); - const channelRef = useRef(channelId); - clientRef.current = client; - channelRef.current = channelId; - useEffect(() => { if (!client || !channelId) { setPresence([]); @@ -80,11 +74,11 @@ export function useChannel(channelId: string | null): UseChannelResult { const sendMessage = useCallback( (payload: unknown) => { - if (clientRef.current && channelRef.current) { - clientRef.current.sendMessage(channelRef.current, payload); + if (client && channelId) { + client?.sendMessage(channelId, payload); } }, - [], + [client, channelId], ); return { presence, messages, sendMessage }; diff --git a/js/src/main.ts b/js/src/main.ts index 007bdf4..51256a5 100644 --- a/js/src/main.ts +++ b/js/src/main.ts @@ -36,6 +36,7 @@ const createWindow = () => { frame: false, webPreferences: { preload: path.join(__dirname, 'preload.js'), + backgroundThrottling: false, }, });