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

@@ -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() {

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)
 

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)
+)

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
 		}

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)
 		}

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

go/k8s/dev/pusher.yaml

@@ -88,7 +88,7 @@ metadata:
   name: pusher-backend-policy
 spec:
   default:
-    timeoutSec: 60
+    timeoutSec: 3600
     connectionDraining:
       drainingTimeoutSec: 30
   targetRef:

go/k8s/prod/pusher.yaml

@@ -85,7 +85,7 @@ metadata:
   name: pusher-backend-policy
 spec:
   default:
-    timeoutSec: 60
+    timeoutSec: 3600
     connectionDraining:
       drainingTimeoutSec: 30
   targetRef:

js/src/hooks/use-channel.ts

@@ -22,12 +22,6 @@ export function useChannel(channelId: string | null): UseChannelResult {
   const [presence, setPresence] = useState<string[]>([]);
   const [messages, setMessages] = useState<ChannelMessage[]>([]);
 
-  // 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 };

js/src/main.ts

@@ -36,6 +36,7 @@ const createWindow = () => {
     frame: false,
     webPreferences: {
       preload: path.join(__dirname, 'preload.js'),
+      backgroundThrottling: false,
     },
   });