diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 1820baa8095ca..a1ace79e43388 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -652,7 +652,11 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { return { kind: "compacting" }; } if (props.activeWorkStartedAt !== null && contentPresentationKind === "ready") { - return { kind: "working", startedAt: props.activeWorkStartedAt }; + return { + kind: "working", + environmentId: props.environmentId, + startedAt: props.activeWorkStartedAt, + }; } if (pendingBackgroundWork !== null && contentPresentationKind === "ready") { return { diff --git a/apps/mobile/src/features/threads/floating-working-control.tsx b/apps/mobile/src/features/threads/floating-working-control.tsx index 24322aa46d43d..a295f69de6922 100644 --- a/apps/mobile/src/features/threads/floating-working-control.tsx +++ b/apps/mobile/src/features/threads/floating-working-control.tsx @@ -1,4 +1,6 @@ +import { toServerClock } from "@t3tools/client-runtime/clock"; import type { SubagentPillSegment } from "@t3tools/client-runtime/state/thread-subagents"; +import type { EnvironmentId } from "@t3tools/contracts"; import { formatDuration } from "@t3tools/shared/orchestrationTiming"; import { GlassContainer, GlassView } from "expo-glass-effect"; import { type ReactNode, useEffect, useRef, useState } from "react"; @@ -511,7 +513,12 @@ function FloatingStatusLabel(props: { ); } return ( - + ); } @@ -550,17 +557,21 @@ function StatusLabelRow(props: { } function WorkingDuration(props: { + readonly environmentId: EnvironmentId; readonly startedAt: string; readonly onLayout: (event: LayoutChangeEvent) => void; }) { return ( - + ); } -export function WorkingTimer(props: { readonly startedAt: string }) { +export function WorkingTimer(props: { + readonly environmentId: EnvironmentId; + readonly startedAt: string; +}) { const nowMs = useVisibleSecondClock(true); return ( - Working {formatWorkingDuration(props.startedAt, nowMs)} + Working {formatWorkingDuration(props.startedAt, toServerClock(props.environmentId, nowMs))} ); } diff --git a/apps/mobile/src/features/threads/floating-working-status.ts b/apps/mobile/src/features/threads/floating-working-status.ts index 264d110f75f74..55868ada184d8 100644 --- a/apps/mobile/src/features/threads/floating-working-status.ts +++ b/apps/mobile/src/features/threads/floating-working-status.ts @@ -1,4 +1,5 @@ import type { EnvironmentConnectionPhase } from "@t3tools/client-runtime/connection"; +import type { EnvironmentId } from "@t3tools/contracts"; export interface FloatingStatusAction { readonly label: string; @@ -8,7 +9,7 @@ export interface FloatingStatusAction { } export type FloatingWorkingStatus = - | { readonly kind: "working"; readonly startedAt: string } + | { readonly kind: "working"; readonly environmentId: EnvironmentId; readonly startedAt: string } | { readonly kind: "syncing"; readonly label: string } | { readonly kind: "compacting" } | { diff --git a/apps/mobile/src/features/threads/thread-list-v2-items.tsx b/apps/mobile/src/features/threads/thread-list-v2-items.tsx index 696c4b7b6c373..3e59ecb3543bc 100644 --- a/apps/mobile/src/features/threads/thread-list-v2-items.tsx +++ b/apps/mobile/src/features/threads/thread-list-v2-items.tsx @@ -25,8 +25,13 @@ import { formatWorkingDurationLabel, resolveThreadWorkingStartedAt, } from "@t3tools/client-runtime/state/models"; -import { AuthOrchestrationOperateScope, type EnvironmentMachineKind } from "@t3tools/contracts"; +import { + AuthOrchestrationOperateScope, + type EnvironmentId, + type EnvironmentMachineKind, +} from "@t3tools/contracts"; import { canSnooze, resolveSnoozePresets } from "@t3tools/client-runtime/state/thread-settled"; +import { toServerClock } from "@t3tools/client-runtime/clock"; import type { MenuAction } from "@react-native-menu/menu"; import { memo, useCallback, useMemo, useState, type ComponentProps } from "react"; import { Pressable, useWindowDimensions, View } from "react-native"; @@ -132,11 +137,14 @@ const DONE_STATUS_LABEL: StatusLabel = { iconTintClassName: "accent-adaptive-emerald-700-300", }; -function ThreadListV2Duration(props: { readonly startedAt: string | null }) { +function ThreadListV2Duration(props: { + readonly environmentId: EnvironmentId; + readonly startedAt: string | null; +}) { const startedAtMs = Date.parse(props.startedAt ?? ""); const nowMs = useVisibleSecondClock(Number.isFinite(startedAtMs)); if (!Number.isFinite(startedAtMs)) return null; - return ` ${formatWorkingDurationLabel(nowMs - startedAtMs)}`; + return ` ${formatWorkingDurationLabel(toServerClock(props.environmentId, nowMs) - startedAtMs)}`; } const CARD_MENU_ACTIONS: MenuAction[] = [ @@ -1084,9 +1092,15 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { > {statusLabel.label} {status === "working" ? ( - + ) : status === "waiting" ? ( - + ) : null} diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index e5696e6f5cd17..61e634de99e27 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1802,6 +1802,8 @@ export const layerWsRpc = ( }), newProjectsRoot: managedFolders.namedProjectsRoot, ...(lastEventLoopStall === undefined ? {} : { lastEventLoopStall }), + // Read last so the stamp is as close to sending as this config gets. + serverTime: DateTime.formatIso(yield* DateTime.now), }; }); diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index 969a5d2e24cc5..138eb39972431 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -1,3 +1,4 @@ +import { recordServerClock, toServerClock } from "@t3tools/client-runtime/clock"; import { boundAttachmentIds, findRecordedWorktreeSetup, @@ -129,6 +130,36 @@ describe("deriveComposerSendState", () => { }); }); +it.each([true, false])( + "counts optimistic sends on the host clock with an existing thread: %s", + (hasThread) => { + const thread = makeThread(); + vi.useFakeTimers(); + vi.setSystemTime(now); + recordServerClock(thread.environmentId, { + serverTimeMs: Date.parse(now) + 3_600_000, + requestedAtMs: Date.parse(now), + receivedAtMs: Date.parse(now), + }); + try { + const dispatch = createLocalDispatchSnapshot(hasThread ? thread : undefined, { + environmentId: thread.environmentId, + }); + expect( + toServerClock(thread.environmentId, Date.parse(now) + 2_000) - + Date.parse(dispatch.startedAt), + ).toBe(2_000); + } finally { + recordServerClock(thread.environmentId, { + serverTimeMs: undefined, + requestedAtMs: 0, + receivedAtMs: 0, + }); + vi.useRealTimers(); + } + }, +); + describe("hasServerAcknowledgedLocalDispatch", () => { it("acknowledges a settled newer background turn", () => { const localDispatch = createLocalDispatchSnapshot( diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index 67c10d9dfeadf..fab28ad215564 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -1,3 +1,4 @@ +import { toServerClock } from "@t3tools/client-runtime/clock"; import { type OrchestrationV2ProviderTurn, type RunAttemptId } from "@t3tools/contracts"; import { deriveLatestThreadRun, @@ -1113,15 +1114,19 @@ export interface LocalDispatchSnapshot { export function createLocalDispatchSnapshot( activeThread: Thread | undefined, options?: { + environmentId?: EnvironmentId; preparingWorktree?: boolean; latestUserMessageId?: ChatMessage["id"] | null; submissionIntent?: ComposerSubmissionIntent; }, ): LocalDispatchSnapshot { + const environmentId = options?.environmentId ?? activeThread?.environmentId; const latestRun = activeThread?.latestRun ?? null; const runtime = activeThread?.runtime ?? null; return { - startedAt: new Date().toISOString(), + startedAt: new Date( + environmentId === undefined ? Date.now() : toServerClock(environmentId, Date.now()), + ).toISOString(), preparingWorktree: Boolean(options?.preparingWorktree), submissionIntent: options?.submissionIntent ?? "foreground", latestUserMessageId: options?.latestUserMessageId ?? null, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 51abfdc64d203..7fe314eb6bf3b 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -1329,6 +1329,7 @@ export default function ChatView(props: ChatViewProps) { isSendBusy, backgroundSubmissionPending, } = useLocalDispatchState({ + environmentId, activeThread, activeLatestRun, latestUserMessageId: diff --git a/apps/web/src/components/chat/TimelineWorkGroupRows.logic.ts b/apps/web/src/components/chat/TimelineWorkGroupRows.logic.ts index e4a8fe4a583c9..bb53e5fa896d1 100644 --- a/apps/web/src/components/chat/TimelineWorkGroupRows.logic.ts +++ b/apps/web/src/components/chat/TimelineWorkGroupRows.logic.ts @@ -1,20 +1,20 @@ +import { toServerClock } from "@t3tools/client-runtime/clock"; +import type { EnvironmentId } from "@t3tools/contracts"; import { formatDuration } from "@t3tools/shared/orchestrationTiming"; -function formatWorkingTimer(startIso: string, endIso: string): string | null { +export function formatWorkingTimerNow(startIso: string, environmentId: EnvironmentId): string { const startedAtMs = Date.parse(startIso); - const endedAtMs = Date.parse(endIso); - if (!Number.isFinite(startedAtMs) || !Number.isFinite(endedAtMs)) { - return null; + if (!Number.isFinite(startedAtMs)) { + return "0s"; } - const elapsedSeconds = Math.max(0, Math.floor((endedAtMs - startedAtMs) / 1000)); + const elapsedSeconds = Math.max( + 0, + Math.floor((toServerClock(environmentId, Date.now()) - startedAtMs) / 1000), + ); if (elapsedSeconds < 60) { return `${elapsedSeconds}s`; } return formatDuration(elapsedSeconds * 1_000); } - -export function formatWorkingTimerNow(startIso: string): string { - return formatWorkingTimer(startIso, new Date().toISOString()) ?? "0s"; -} diff --git a/apps/web/src/components/chat/TimelineWorkGroupRows.tsx b/apps/web/src/components/chat/TimelineWorkGroupRows.tsx index c122caa96f32a..a072a47ef5430 100644 --- a/apps/web/src/components/chat/TimelineWorkGroupRows.tsx +++ b/apps/web/src/components/chat/TimelineWorkGroupRows.tsx @@ -110,9 +110,11 @@ export const WorkGroupSection = memo(function WorkGroupSection({ }); function WorkingTimer({ createdAt }: { createdAt: string }) { + const { activeThreadEnvironmentId } = use(TimelineRowCtx); + return ( formatWorkingTimerNow(createdAt), true)} + ref={tickTextContent(() => formatWorkingTimerNow(createdAt, activeThreadEnvironmentId), true)} className="tabular-nums" /> ); diff --git a/apps/web/src/components/chat/useLocalDispatchState.ts b/apps/web/src/components/chat/useLocalDispatchState.ts index 977028db17b9e..eee86c699d8c5 100644 --- a/apps/web/src/components/chat/useLocalDispatchState.ts +++ b/apps/web/src/components/chat/useLocalDispatchState.ts @@ -1,4 +1,4 @@ -import { type MessageId, type RuntimeRequestId } from "@t3tools/contracts"; +import { type EnvironmentId, type MessageId, type RuntimeRequestId } from "@t3tools/contracts"; import { useState } from "react"; import { type ComposerSubmissionIntent } from "../../composer-logic"; import { type SessionPhase, type Thread } from "../../types"; @@ -9,6 +9,7 @@ import { } from "../ChatView.logic"; export function useLocalDispatchState(input: { + environmentId: EnvironmentId; activeThread: Thread | undefined; activeLatestRun: Thread["latestRun"] | null; latestUserMessageId: MessageId | null; @@ -50,6 +51,7 @@ export function useLocalDispatchState(input: { } return createLocalDispatchSnapshot(input.activeThread, { ...options, + environmentId: input.environmentId, latestUserMessageId: input.latestUserMessageId, }); }); diff --git a/apps/web/src/components/sidebar/SidebarThreadRow.tsx b/apps/web/src/components/sidebar/SidebarThreadRow.tsx index 8837d47ae0eec..c676e1d1b0d18 100644 --- a/apps/web/src/components/sidebar/SidebarThreadRow.tsx +++ b/apps/web/src/components/sidebar/SidebarThreadRow.tsx @@ -20,8 +20,10 @@ import { resolveThreadProviderStack, } from "@t3tools/client-runtime/state/models"; import { scopeThreadRef, scopedThreadKey } from "@t3tools/client-runtime/environment"; +import { toServerClock } from "@t3tools/client-runtime/clock"; import { AuthOrchestrationOperateScope, + type EnvironmentId, type EnvironmentMachineKind, type ScopedThreadRef, } from "@t3tools/contracts"; @@ -158,18 +160,20 @@ function JumpHintBadge(props: { label: string }) { ); } -function WorkingDuration(props: { startedAt: string | null }) { +function WorkingDuration(props: { environmentId: EnvironmentId; startedAt: string | null }) { const startedMs = props.startedAt !== null ? Date.parse(props.startedAt) : Number.NaN; if (Number.isNaN(startedMs)) return null; - return ; + return ; } // Self-ticking so only this span re-renders each second, not the whole row; // the shared tick lands every working row's update in the same frame. -function WorkingDurationLabel(props: { startedMs: number }) { +function WorkingDurationLabel(props: { environmentId: EnvironmentId; startedMs: number }) { const nowMs = useNowSecond(); return ( - {formatWorkingDurationLabel(nowMs - props.startedMs)} + + {formatWorkingDurationLabel(toServerClock(props.environmentId, nowMs) - props.startedMs)} + ); } @@ -1386,11 +1390,17 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP {topStatus.label} {status === "working" ? ( - + ) : status === "waiting" ? ( - + ) : null} diff --git a/packages/client-runtime/src/clock.test.ts b/packages/client-runtime/src/clock.test.ts index ca8130a379ad4..cc12e9bb7ec30 100644 --- a/packages/client-runtime/src/clock.test.ts +++ b/packages/client-runtime/src/clock.test.ts @@ -1,5 +1,6 @@ +import { EnvironmentId } from "@t3tools/contracts"; import { afterEach, describe, expect, it, vi } from "vite-plus/test"; -import { deadlineStore, secondTick } from "./clock.ts"; +import { deadlineStore, recordServerClock, secondTick, toServerClock } from "./clock.ts"; describe("deadlineStore", () => { afterEach(() => { @@ -138,3 +139,66 @@ describe("secondTick", () => { expect(vi.getTimerCount()).toBe(0); }); }); + +describe("server clock", () => { + const HOUR = 3_600_000; + + it("counts from a host clock running ahead, minus only the reply's latency", () => { + const environmentId = EnvironmentId.make("ahead"); + // Stamped at the host's 1h + 1_000 while this clock read about 1_000. + recordServerClock(environmentId, { + serverTimeMs: HOUR + 1_000, + requestedAtMs: 900, + receivedAtMs: 1_050, + }); + expect(toServerClock(environmentId, 2_050)).toBe(HOUR + 2_000); + }); + + it("counts from a host clock running behind, minus only the request's latency", () => { + const environmentId = EnvironmentId.make("behind"); + // Stamped at the host's 1_000 while this clock read about 1h + 1_000. + recordServerClock(environmentId, { + serverTimeMs: 1_000, + requestedAtMs: HOUR + 900, + receivedAtMs: HOUR + 1_050, + }); + expect(toServerClock(environmentId, HOUR + 1_900)).toBe(2_000); + }); + + it("leaves clocks that agree within the round trip unshifted", () => { + const environmentId = EnvironmentId.make("synced"); + recordServerClock(environmentId, { + serverTimeMs: 5_000, + requestedAtMs: 1_000, + receivedAtMs: 9_000, + }); + expect(toServerClock(environmentId, 12_345)).toBe(12_345); + }); + + it("replaces an environment's reading on reconnect", () => { + const environmentId = EnvironmentId.make("reconnected"); + recordServerClock(environmentId, { serverTimeMs: HOUR, requestedAtMs: 0, receivedAtMs: 0 }); + recordServerClock(environmentId, { serverTimeMs: 0, requestedAtMs: 0, receivedAtMs: 0 }); + expect(toServerClock(environmentId, 500)).toBe(500); + }); + + it("keeps readings separate for each environment", () => { + const ahead = EnvironmentId.make("isolated-ahead"); + const behind = EnvironmentId.make("isolated-behind"); + recordServerClock(ahead, { serverTimeMs: HOUR, requestedAtMs: 0, receivedAtMs: 0 }); + recordServerClock(behind, { serverTimeMs: 0, requestedAtMs: HOUR, receivedAtMs: HOUR }); + expect(toServerClock(ahead, 500)).toBe(HOUR + 500); + expect(toServerClock(behind, HOUR + 500)).toBe(500); + expect(toServerClock(EnvironmentId.make("unknown"), 500)).toBe(500); + }); + + it.each([undefined, Number.NaN, Number.POSITIVE_INFINITY])( + "drops the previous reading when the next stamp is %s", + (serverTimeMs) => { + const environmentId = EnvironmentId.make("missing-stamp"); + recordServerClock(environmentId, { serverTimeMs: HOUR, requestedAtMs: 0, receivedAtMs: 0 }); + recordServerClock(environmentId, { serverTimeMs, requestedAtMs: 0, receivedAtMs: 0 }); + expect(toServerClock(environmentId, 500)).toBe(500); + }, + ); +}); diff --git a/packages/client-runtime/src/clock.ts b/packages/client-runtime/src/clock.ts index d0e9db7259530..f8f1290730084 100644 --- a/packages/client-runtime/src/clock.ts +++ b/packages/client-runtime/src/clock.ts @@ -1,5 +1,7 @@ // @effect-diagnostics globalTimers:off globalDate:off - Render clocks for React hooks, outside an Effect runtime. +import type { EnvironmentId } from "@t3tools/contracts"; + /** * A `useSyncExternalStore`-shaped clock. Snapshots stay cached between * notifications while subscribed, as `useSyncExternalStore` requires. @@ -70,6 +72,43 @@ export const minuteTick: ClockStore = makeTickStore( 60_000, ); +// How far each environment's clock runs ahead of this one. Timers read it on +// every tick, so a reconnect's new reading lands without a re-render. +const serverClockOffsets = new Map(); + +/** + * The host stamped `serverTimeMs` between the request and its reply, so the + * offset lies within that window. Keep the one nearest 0, so clocks that agree + * read as 0 instead of shifting every timer by network latency. + */ +export function recordServerClock( + environmentId: EnvironmentId, + sample: { + readonly serverTimeMs: number | undefined; + readonly requestedAtMs: number; + readonly receivedAtMs: number; + }, +): void { + const { serverTimeMs, requestedAtMs, receivedAtMs } = sample; + if (serverTimeMs === undefined || !Number.isFinite(serverTimeMs)) { + serverClockOffsets.delete(environmentId); + return; + } + serverClockOffsets.set( + environmentId, + serverTimeMs > receivedAtMs + ? serverTimeMs - receivedAtMs + : serverTimeMs < requestedAtMs + ? serverTimeMs - requestedAtMs + : 0, + ); +} + +/** Converts client epoch milliseconds to the environment's clock. */ +export function toServerClock(environmentId: EnvironmentId, clientMs: number): number { + return clientMs + (serverClockOffsets.get(environmentId) ?? 0); +} + export interface DeadlineStore extends ClockStore { /** * Re-checks `Date.now()` and notifies if the deadline has passed. For callers diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index b0c5857ca9caf..3c556349ede6c 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -21,6 +21,7 @@ import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as Socket from "effect/socket/Socket"; +import * as TestClock from "effect/testing/TestClock"; import { AVAILABLE_CONNECTION_STATE, @@ -32,6 +33,7 @@ import { import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as Persistence from "../platform/persistence.ts"; import * as RpcSession from "./session.ts"; +import { toServerClock } from "../clock.ts"; import { makeEnvironmentServerConfigState } from "../state/server.ts"; import { applyServerConfigProjection } from "../state/serverConfigProjection.ts"; @@ -426,6 +428,59 @@ describe("RpcSessionFactory", () => { ), ); + it.effect("reads the host clock from the first config snapshot only", () => + Effect.scoped( + Effect.gen(function* () { + const hourAheadConfig = { ...SERVER_CONFIG, serverTime: "1970-01-01T01:00:00.000Z" }; + const { factory, sockets } = yield* makeFactory(); + const session = yield* factory.connect(PREPARED); + const ready = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket, encodeServerConfig(hourAheadConfig)); + yield* Fiber.join(ready); + expect(toServerClock(PREPARED.environmentId, 0)).toBe(3_600_000); + + // A later snapshot, such as late editor discovery, repeats the old stamp. + yield* TestClock.adjust("1 second"); + const observed = yield* Queue.unbounded(); + yield* session.subscribeServerConfig({}).pipe( + Stream.runForEach((event) => Queue.offer(observed, event)), + Effect.forkChild, + ); + yield* Queue.take(observed); + yield* publishConfigEvents(socket, [ + { version: 1, type: "snapshot", config: hourAheadConfig }, + ]); + yield* Queue.take(observed); + expect(toServerClock(PREPARED.environmentId, 0)).toBe(3_600_000); + }), + ), + ); + + it.effect("drops the old host clock when a reconnect's snapshot has none", () => + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(); + const connectWith = (config: unknown) => + Effect.scoped( + Effect.gen(function* () { + const socketIndex = sockets.length; + const session = yield* factory.connect(PREPARED); + const ready = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets, socketIndex); + socket.open(); + yield* completeInitialConfig(socket, config); + yield* Fiber.join(ready); + }), + ); + + yield* connectWith({ ...ENCODED_SERVER_CONFIG, serverTime: "1970-01-01T01:00:00.000Z" }); + expect(toServerClock(PREPARED.environmentId, 0)).toBe(3_600_000); + yield* connectWith(ENCODED_SERVER_CONFIG); + expect(toServerClock(PREPARED.environmentId, 0)).toBe(0); + }), + ); + it.effect.each([ { usageLimitSources: true }, { environmentThemes: true, usageLimitSources: true }, diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index ebd05d0b2d301..22442e8cc801d 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -5,6 +5,7 @@ import { WS_METHODS, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; @@ -42,6 +43,7 @@ import { withoutEnvironmentThemes, } from "../state/serverConfigProjection.ts"; import { environmentMismatchError } from "../connection/errors.ts"; +import { recordServerClock } from "../clock.ts"; const SOCKET_OPEN_TIMEOUT = "15 seconds"; @@ -252,6 +254,7 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( reason: "remote-unavailable", detail: `${connection.label} config subscription ended.`, }); + const configRequestedAtMs = yield* Clock.currentTimeMillis; const serverConfigSource = protocolClient[WS_METHODS.subscribeServerConfig]( onDemandSnapshots(serverConfigInput), ).pipe( @@ -295,6 +298,16 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( yield* PubSub.publish(serverConfigUpdates, buffered.value); } if (event.type === "snapshot") { + const { serverTime } = event.config; + // Later snapshots repeat the first one's stale serverTime. Record + // before readiness so the first timers already count on its clock. + if (!(yield* Deferred.isDone(initialConfigDeferred))) { + recordServerClock(connection.environmentId, { + serverTimeMs: serverTime === undefined ? undefined : Date.parse(serverTime), + requestedAtMs: configRequestedAtMs, + receivedAtMs: yield* Clock.currentTimeMillis, + }); + } yield* Deferred.succeed(initialConfigDeferred, event.config); } }), diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index f1c8d3c19e83d..d7628becb3353 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -428,11 +428,13 @@ export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConf const persist = Effect.fn("EnvironmentServerConfigState.persist")(function* ( config: ServerConfig, ) { - // A cached stall would read as a fresh freeze on the next launch. + // A cached stall would read as a fresh freeze on the next launch, and + // serverTime changes every connect, which would rewrite the cache each time. const { environmentThemes: _themes, usageLimitSources: _sources, lastEventLoopStall: _stall, + serverTime: _serverTime, ...cacheable } = config; if (lastPersisted !== undefined && sameCacheValue(lastPersisted, cacheable)) { diff --git a/packages/contracts/src/server.ts b/packages/contracts/src/server.ts index c14b4b4b24eaf..390b75b6fe188 100644 --- a/packages/contracts/src/server.ts +++ b/packages/contracts/src/server.ts @@ -620,6 +620,12 @@ export const ServerConfig = Schema.Struct({ * answering. Clients do not cache it, since it describes the host at connect. */ lastEventLoopStall: Schema.optionalKey(ServerEventLoopStall), + /** + * The host's clock when it built the config, so clients can count live + * elapsed times against host timestamps despite clock skew. Only a + * subscription's first snapshot is fresh; later snapshots repeat it. + */ + serverTime: Schema.optionalKey(IsoDateTime), /** * Never sent in a config snapshot: the theme stream emits the current set * before any change, so a snapshot carrying it too would hand every