-
Notifications
You must be signed in to change notification settings - Fork 12
Mobile + desktop enhancements #216
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. Weβll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -100,6 +100,7 @@ const DEFAULT_SYNC_HEARTBEAT_INTERVAL_MS = 30_000; | |
| const DEFAULT_SYNC_POLL_INTERVAL_MS = 400; | ||
| const DEFAULT_BRAIN_STATUS_INTERVAL_MS = 5_000; | ||
| const DEFAULT_TERMINAL_SNAPSHOT_BYTES = 220_000; | ||
| const PEER_BACKPRESSURE_BYTES = 4 * 1024 * 1024; | ||
| const LANE_PRESENCE_TTL_MS = 60_000; | ||
| const SYNC_MDNS_SERVICE_TYPE = "ade-sync"; | ||
| export const SYNC_TAILNET_DISCOVERY_SERVICE_NAME = "svc:ade-sync"; | ||
|
|
@@ -753,6 +754,13 @@ export function createSyncHostService(args: SyncHostServiceArgs) { | |
| const sentAt = nowIso(); | ||
| for (const peer of peers) { | ||
| if (!peer.authenticated || peer.ws.readyState !== WebSocket.OPEN) continue; | ||
| if (isPeerBackpressured(peer)) { | ||
| args.logger.debug("sync_host.heartbeat_deferred_backpressure", { | ||
| peerDeviceId: peer.metadata?.deviceId ?? null, | ||
| bufferedAmount: peer.ws.bufferedAmount, | ||
| }); | ||
| continue; | ||
| } | ||
| if (peer.awaitingHeartbeatAt) { | ||
| peer.missedHeartbeatCount += 1; | ||
| if (peer.missedHeartbeatCount >= 2) { | ||
|
|
@@ -1096,9 +1104,14 @@ export function createSyncHostService(args: SyncHostServiceArgs) { | |
| }; | ||
|
|
||
| function send<TPayload>(ws: WebSocket, type: SyncEnvelope["type"], payload: TPayload, requestId?: string | null): void { | ||
| if (ws.readyState !== WebSocket.OPEN) return; | ||
| ws.send(encodeSyncEnvelope({ type, payload, requestId, compressionThresholdBytes })); | ||
| } | ||
|
|
||
| function isPeerBackpressured(peer: PeerState): boolean { | ||
| return peer.ws.bufferedAmount >= PEER_BACKPRESSURE_BYTES; | ||
| } | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Backpressure policy is not centralized, so ungated sends can still grow buffers.
π‘οΈ Proposed fix (centralized enforcement with opt-out)-function send<TPayload>(ws: WebSocket, type: SyncEnvelope["type"], payload: TPayload, requestId?: string | null): void {
- if (ws.readyState !== WebSocket.OPEN) return;
- ws.send(encodeSyncEnvelope({ type, payload, requestId, compressionThresholdBytes }));
+function send<TPayload>(
+ peer: PeerState,
+ type: SyncEnvelope["type"],
+ payload: TPayload,
+ requestId?: string | null,
+ options?: { force?: boolean },
+): void {
+ if (peer.ws.readyState !== WebSocket.OPEN) return;
+ if (!options?.force && isPeerBackpressured(peer)) return;
+ peer.ws.send(encodeSyncEnvelope({ type, payload, requestId, compressionThresholdBytes }));
}π€ Prompt for AI Agents |
||
|
|
||
| function sendAndWait<TPayload>( | ||
| ws: WebSocket, | ||
| type: SyncEnvelope["type"], | ||
|
|
@@ -1421,6 +1434,7 @@ export function createSyncHostService(args: SyncHostServiceArgs) { | |
|
|
||
| for (const peer of peers) { | ||
| if (!peer.authenticated || peer.ws.readyState !== WebSocket.OPEN) continue; | ||
| if (isPeerBackpressured(peer)) continue; | ||
| for (const sessionId of peer.subscribedChatSessionIds) { | ||
| const session = args.sessionService.get(sessionId); | ||
| if (!session?.transcriptPath) continue; | ||
|
|
@@ -1441,6 +1455,7 @@ export function createSyncHostService(args: SyncHostServiceArgs) { | |
| function broadcastChatEvent(event: AgentChatEventEnvelope): void { | ||
| for (const peer of peers) { | ||
| if (!peer.authenticated || peer.ws.readyState !== WebSocket.OPEN) continue; | ||
| if (isPeerBackpressured(peer)) continue; | ||
| if (!peer.subscribedChatSessionIds.has(event.sessionId)) continue; | ||
| if (!rememberChatEventSent(peer, event)) continue; | ||
| send(peer.ws, "chat_event", event); | ||
|
|
@@ -1452,6 +1467,7 @@ export function createSyncHostService(args: SyncHostServiceArgs) { | |
| const currentDbVersion = args.db.sync.getDbVersion(); | ||
| for (const peer of peers) { | ||
| if (!peer.authenticated || !peer.metadata || peer.ws.readyState !== WebSocket.OPEN) continue; | ||
| if (isPeerBackpressured(peer)) continue; | ||
| if (currentDbVersion <= peer.lastKnownServerDbVersion) continue; | ||
| const changes = args.db.sync | ||
| .exportChangesSince(peer.lastKnownServerDbVersion) | ||
|
|
@@ -2394,6 +2410,7 @@ export function createSyncHostService(args: SyncHostServiceArgs) { | |
| }; | ||
| for (const peer of peers) { | ||
| if (!peer.authenticated || !peer.subscribedSessionIds.has(event.sessionId) || peer.ws.readyState !== WebSocket.OPEN) continue; | ||
| if (isPeerBackpressured(peer)) continue; | ||
| send(peer.ws, "terminal_data", payload); | ||
| } | ||
| }, | ||
|
|
@@ -2407,6 +2424,7 @@ export function createSyncHostService(args: SyncHostServiceArgs) { | |
| }; | ||
| for (const peer of peers) { | ||
| if (!peer.authenticated || !peer.subscribedSessionIds.has(event.sessionId) || peer.ws.readyState !== WebSocket.OPEN) continue; | ||
| if (isPeerBackpressured(peer)) continue; | ||
| send(peer.ws, "terminal_exit", payload); | ||
| } | ||
| }, | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.