feat(sync-service): host multiplexed sync on a dedicated DSS socket

.github/workspace-dep-closures.json

@@ -5334,6 +5334,7 @@
       "crates/sqs_worker",
       "crates/static_file",
       "crates/static_file_service_client",
+      "crates/sync_router",
       "crates/sync_service",
       "crates/sync_service_client",
       "crates/system_properties",
@@ -5349,7 +5350,8 @@
       "crates/whatsapp_linked",
       "crates/work_feed",
       "crates/workspace-hack",
-      "services/document_storage_service"
+      "services/document_storage_service",
+      "services/sync-service/bebop-schema"
     ],
     "document_storage_service_client": [
       "crates/agent",
@@ -13289,15 +13291,8 @@
       "crates/workspace-hack"
     ],
     "sync_router": [
-      "crates/connection_gateway_models",
-      "crates/cowlike",
-      "crates/macro_entrypoint",
-      "crates/macro_env",
-      "crates/macro_env_var",
-      "crates/macro_user_id",
-      "crates/model-entity",
+      "crates/sync_router",
       "crates/workspace-hack",
-      "services/sync-router",
       "services/sync-service/bebop-schema"
     ],
     "sync_service": [

Cargo.toml

@@ -176,7 +176,7 @@ members = [
   "services/sha_cleanup_worker",
   "services/slack_import_worker",
   "services/static_file_service",
-  "services/sync-router",
+  "crates/sync_router",
   "services/sync-service",
   "services/sync-service/bebop-schema",
   "services/unfurl_service",

apps/web/AGENTS.md

@@ -127,12 +127,13 @@ Then trigger the interaction and read `window.__inst.log`. `'1,2,3' → '' → '
 
 Document sync uses the `sync-via-gateway` PostHog flag in both deployed dev and
 production. Off or unknown flag values use direct sync sockets.
-Local stacks with `sync_router` can opt in with `VITE_SYNC_VIA_GATEWAY=true`;
+Local stacks with the updated DSS can opt in with `VITE_SYNC_VIA_GATEWAY=true`;
 `false` forces direct sync. The choice is made when a sync source is created
 and stays fixed for its lifetime. PostHog can serve a cached value on reload;
 recreate document sessions after the client receives the updated flag to change
-transport. This is not an immediate kill switch for open sessions. Provision
-the hosted router before enabling.
+transport. This is not an immediate kill switch for open sessions. Finish deploying the updated
+DSS before enabling; the dedicated `/sync/ws` socket hosts routing independently
+of the connection-gateway notification socket. The PostHog key keeps its historical name.
 
 Saved canvases use the shared Loro runtime; the stored JSON is only the original
 migration seed. Keep whole-file simple-save out of scene editing. Canvas

apps/web/src/lib/core/constant/featureFlags.ts

@@ -216,8 +216,8 @@ export const ENABLE_SCRIPTING = defineFlag({
   default: false,
 }).enabled;
 
-// Deployed dev and production defer to PostHog: the router must be provisioned
-// before enabling gateway sync. Local stacks can use the env override.
+// Historical key retained at the deployed PostHog boundary. This selects the
+// dedicated DSS sync socket; deploy DSS before enabling. Local env override supported.
 export const syncViaGateway = defineFlag({
   key: 'sync-via-gateway',
   env: 'SYNC_VIA_GATEWAY',

apps/web/src/lib/service-clients/service-sync/source/helpers.test.ts

@@ -23,8 +23,8 @@ vi.mock('@core/constant/featureFlags', () => ({
   syncViaGateway: { key: 'sync-via-gateway', override: undefined },
   isFeatureEnabled: () => false,
 }));
-vi.mock('@service-connection/websocket-url', () => ({
-  resolveWsUrl: () => 'ws://gateway.test',
+vi.mock('../websocket-url', () => ({
+  resolveSyncRouterWsUrl: () => 'ws://dss.test/sync/ws',
 }));
 vi.mock('@service-storage/client', () => ({ storageServiceClient: {} }));
 

apps/web/src/lib/service-clients/service-sync/source/helpers.ts

@@ -7,7 +7,7 @@ import type {
   SyncError,
   TimeoutError,
 } from '@macro-inc/collaboration/collab/source';
-import { GatewaySyncTransport } from '@macro-inc/collaboration/sync-service/gateway';
+import { SyncRouterTransport } from '@macro-inc/collaboration/sync-service/router';
 import {
   createSyncSocket,
   type SyncSocket,
@@ -19,9 +19,9 @@ import {
 } from '@macro-inc/collaboration/sync-service/source';
 import type { UrlResolver } from '@macro-inc/collaboration/websocket';
 import { createWebsocketStateSignal } from '@macro-inc/collaboration/websocket/solid/state-signal';
-import { resolveWsUrl as resolveGatewayWsUrl } from '@service-connection/websocket-url';
 import { storageServiceClient } from '@service-storage/client';
 import type { ResultAsync } from 'neverthrow';
+import { resolveSyncRouterWsUrl } from '../websocket-url';
 import type { DocumentSyncAuthorization } from './authorization';
 import { guardDocumentSyncSource } from './guarded-source';
 
@@ -107,10 +107,10 @@ export function createSyncServiceSocket(
   );
 }
 
-let gatewayTransport: GatewaySyncTransport | undefined;
-function syncGatewayTransport(): GatewaySyncTransport {
-  gatewayTransport ??= new GatewaySyncTransport(resolveGatewayWsUrl);
-  return gatewayTransport;
+let routerTransport: SyncRouterTransport | undefined;
+function syncRouterTransport(): SyncRouterTransport {
+  routerTransport ??= new SyncRouterTransport(resolveSyncRouterWsUrl);
+  return routerTransport;
 }
 
 async function refreshPermissionToken(documentId: string): Promise<string> {
@@ -133,7 +133,7 @@ export const createSyncServiceSource = (
   // Pick a transport for this source's lifetime. Later flag changes apply to
   // newly created sources without interrupting an editor's active sync session.
   const ws: SyncSocket = isFeatureEnabled(syncViaGateway)
-    ? syncGatewayTransport().attach(
+    ? syncRouterTransport().attach(
         documentId,
         authorization ? undefined : token,
         authorization?.getToken ?? (() => refreshPermissionToken(documentId))

apps/web/src/lib/service-clients/service-sync/websocket-url.test.ts

@@ -0,0 +1,52 @@
+import { err, ok } from 'neverthrow';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+const auth = vi.hoisted(() => ({
+  bearer: false,
+  token: vi.fn(async () => 'token+/='),
+  cookie: vi.fn(),
+}));
+vi.mock('@core/constant/featureFlags', () => ({
+  get ENABLE_BEARER_TOKEN_AUTH() {
+    return auth.bearer;
+  },
+}));
+vi.mock('@core/constant/servers', () => ({
+  SERVER_HOSTS: { 'document-storage-service': 'https://gateway.test/dss' },
+}));
+vi.mock('@core/util/fetchWithToken', () => ({ fetchToken: auth.cookie }));
+vi.mock('@service-auth/fetch', () => ({ getMacroApiToken: auth.token }));
+
+import { resolveSyncRouterWsUrl } from './websocket-url';
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  auth.bearer = false;
+  auth.cookie.mockResolvedValue(ok(undefined));
+});
+
+describe('dedicated DSS sync socket credentials', () => {
+  it('preserves the DSS gateway prefix and refreshes cookies on each connection', async () => {
+    expect(await resolveSyncRouterWsUrl()).toBe(
+      'wss://gateway.test/dss/sync/ws'
+    );
+    await resolveSyncRouterWsUrl();
+    expect(auth.cookie).toHaveBeenCalledTimes(2);
+    expect(auth.token).not.toHaveBeenCalled();
+  });
+
+  it('encodes a fresh bearer token in the upgrade query', async () => {
+    auth.bearer = true;
+    const url = new URL(await resolveSyncRouterWsUrl());
+    expect(url.pathname).toBe('/dss/sync/ws');
+    expect(url.searchParams.get('macro-api-token')).toBe('token+/=');
+    expect(auth.cookie).not.toHaveBeenCalled();
+  });
+
+  it('does not connect after a failed cookie refresh', async () => {
+    auth.cookie.mockResolvedValue(err([{ code: 'UNAUTHORIZED' }]));
+    await expect(resolveSyncRouterWsUrl()).rejects.toThrow(
+      'Unable to refresh sync session'
+    );
+  });
+});

apps/web/src/lib/service-clients/service-sync/websocket-url.ts

@@ -0,0 +1,26 @@
+import { ENABLE_BEARER_TOKEN_AUTH } from '@core/constant/featureFlags';
+import { SERVER_HOSTS } from '@core/constant/servers';
+import { fetchToken } from '@core/util/fetchWithToken';
+import { Telemetry } from '@macro-inc/observability';
+import { getMacroApiToken } from '@service-auth/fetch';
+
+/** Resolve fresh account credentials for the dedicated DSS sync socket. */
+export async function resolveSyncRouterWsUrl(): Promise<string> {
+  return Telemetry.span('sync.router.connect', async (span) => {
+    const url = new URL(
+      `${SERVER_HOSTS['document-storage-service'].replace(/\/$/, '')}/sync/ws`
+    );
+    url.protocol = url.protocol === 'http:' ? 'ws:' : 'wss:';
+    if (ENABLE_BEARER_TOKEN_AUTH) {
+      const token = await getMacroApiToken();
+      if (!token) throw new Error('No Macro API token');
+      url.searchParams.set('macro-api-token', token);
+    } else {
+      const refreshed = await fetchToken();
+      if (refreshed.isErr()) throw new Error('Unable to refresh sync session');
+    }
+    const traceparent = span.traceparent();
+    if (traceparent) url.searchParams.set('traceparent', traceparent);
+    return url.toString();
+  });
+}

crates/sync_router/AGENTS.md

@@ -0,0 +1,41 @@
+# sync_router
+
+## What it is
+
+Document sync routing library hosted by DSS at `/sync/ws`. Each authenticated
+WebSocket owns a router and its downstream connections; there is no separate
+service, Redis subscriber, or Pulumi stack.
+
+## Start here
+
+- `src/domain/router.rs`: document subscriptions, bounded queues, route epochs.
+- `src/domain/ports.rs`: socket delivery and downstream lifecycle contracts.
+- `src/outbound/do_downstream.rs`: scoped-token WebSocket dialing and frame pumps.
+- `src/session.rs`: socket-owned task
+  tracking, cancellation, and local delivery adapter.
+
+## Scope
+
+- Owns routing and cancellation, not document state or authorization.
+- `services/sync-service` validates each document-scoped token and owns CRDT state.
+- Browser envelopes are generated from `services/sync-service/bebop/schema.bop`.
+
+## Checks
+
+- `cargo test -p sync_router` (no live external services required).
+- Session tests exercise real local downstream sockets, isolation, backpressure,
+  cancellation, and byte budgets.
+- After dependency changes, regenerate and check `cargo x deps`.
+
+## Gotchas
+
+- Preserve inner chunk framing byte-for-byte. A full queue must close its route
+  or socket and trigger resync; never drop a chunk and continue the stream.
+- Inbound messages are capped at one MiB; downstream queues share an eight MiB
+  budget per session, retained through the write. Never increase count limits
+  without checking the byte bound.
+- Dropping a downstream route cancels its task, including pending dials/writes.
+- Every DSS replica handles only its own sockets. No distributed route
+  ownership or singleton deployment is required.
+- Deploy the updated DSS before enabling PostHog `sync-via-gateway`.
+  `VITE_SYNC_VIA_GATEWAY=true` opts local clients in.

services/sync-router/Cargo.toml → crates/sync_router/Cargo.toml

@@ -7,20 +7,17 @@ version = "0.1.0"
 [dependencies]
 anyhow = { workspace = true }
 bebop = { workspace = true }
-connection_gateway_models = { path = "../../crates/connection_gateway_models" }
 futures = { workspace = true }
-macro_entrypoint = { path = "../../crates/macro_entrypoint" }
-macro_env_var = { path = "../../crates/macro_env_var" }
 opentelemetry = { workspace = true }
-postcard = { workspace = true }
-redis = { workspace = true, features = ["aio", "tokio-comp", "tokio-rustls-comp"] }
-sync_service_bebop_schema = { path = "../sync-service/bebop-schema", features = ["bebop-owned-all"] }
+sync_service_bebop_schema = { path = "../../services/sync-service/bebop-schema", features = ["bebop-owned-all"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] }
+tokio-util = { workspace = true, features = ["rt"] }
+url = { workspace = true }
 tokio-tungstenite = { workspace = true }
 tracing = { workspace = true }
 tracing-opentelemetry = { workspace = true }
-workspace-hack = { version = "0.1", path = "../../crates/workspace-hack" }
+workspace-hack = { version = "0.1", path = "../workspace-hack" }
 
 [dev-dependencies]
 mockall = { workspace = true }