perf(session): send one projection frame per committed event (#7587)
* perf(session): send one projection frame per committed event * fix(session): deliver keys changed before a drive failure; address review * perf(session): carry projection frame changes as key and val entries
This commit is contained in:
1 parent
76bb45e2be
commit
945006c984
51 files changed
+631
-283
No files matched your search
@@ -642,15 +642,17 @@ export class SessionManager {
|
||||
// ---- Live control and Host-event sinks ----
|
||||
|
||||
/**
|
||||
* Apply a complete control baseline or one later replacement frame.
|
||||
* @param frame - baseline or live control replacement from Session Controller.
|
||||
* Apply a complete control baseline or one later per-event projection frame.
|
||||
* @param frame - baseline or live per-event projection change from Session Controller.
|
||||
*/
|
||||
handleControlFrame(frame: SessionControlFrame): void {
|
||||
if (frame.type === 'baseline') {
|
||||
this.replaceControlBaseline(frame.value)
|
||||
return
|
||||
}
|
||||
this.projectionStore(frame.sessionId).apply(frame.key, frame.value, SessionSeq(frame.seq))
|
||||
const store = this.projectionStore(frame.sessionId)
|
||||
const seq = SessionSeq(frame.seq)
|
||||
for (const change of frame.changes) store.apply(change.key, change.val, seq)
|
||||
this.notifier.markDirty()
|
||||
}
|
||||
|
||||
|
||||
@@ -5,11 +5,11 @@ import { Deque } from '@deepseek-ai/dsh-deque'
|
||||
import type {
|
||||
Session, SessionId,
|
||||
} from '@deepseek-ai/dsh-session'
|
||||
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
|
||||
import type {
|
||||
SessionControlBaseline,
|
||||
SessionControlFrame,
|
||||
SessionProjectionBaseline,
|
||||
SessionProjectionUpdate,
|
||||
SessionProjectionValues,
|
||||
} from './types.ts'
|
||||
|
||||
@@ -19,13 +19,13 @@ export class SessionControlController {
|
||||
|
||||
/** @param ctx - Host context carrying live Agent and projection services. */
|
||||
constructor(private readonly ctx: Context) {
|
||||
ctx.sessionProjections.onChanged((session, key, value, seq) => {
|
||||
ctx.sessionProjections.onEventChanged((session, changes, seq) => {
|
||||
this.broadcast({
|
||||
type: 'projection',
|
||||
sessionId: session.id,
|
||||
key,
|
||||
value: value as JsonValue,
|
||||
seq,
|
||||
// Every projection definition validates its value before change publication.
|
||||
changes: changes as SessionProjectionUpdate['changes'],
|
||||
})
|
||||
})
|
||||
ctx.effect(() => () => {
|
||||
|
||||
@@ -103,10 +103,10 @@ export class HistoryController {
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe to sparse projection replacements without reading a baseline.
|
||||
* Subscribe to projection changes without reading a baseline.
|
||||
* @param request - Session identity to subscribe without reading a baseline.
|
||||
* @param signal - stream cancellation.
|
||||
* @returns readiness followed by sparse per-key replacement values.
|
||||
* @returns readiness followed by one change frame per committed event, listing every key it changed.
|
||||
*/
|
||||
async *observeProjections(request: SessionProjectionsRequest, signal: AbortSignal): AsyncIterable<SessionProjectionFrame> {
|
||||
using listener = new ProjectionListener(this.ctx, request.sessionId, signal)
|
||||
|
||||
@@ -5,7 +5,6 @@ import { Deque } from '@deepseek-ai/dsh-deque'
|
||||
import { SessionLogOffset } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
|
||||
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
|
||||
import { entryFor } from '../history.ts'
|
||||
import type { SessionProjectionChange, SessionEventEntry } from '../types.ts'
|
||||
|
||||
@@ -128,7 +127,7 @@ export function eventItem(item: EventItem, next?: number): {
|
||||
return { events, next }
|
||||
}
|
||||
|
||||
/** Sparse projection changes retain every key, including keys sharing a seq. */
|
||||
/** Projection changes of one Session, one change per committed event that changed a projection. */
|
||||
export class ProjectionListener implements Disposable {
|
||||
private readonly buffer: ListenerBuffer<{ session: Session; change: SessionProjectionChange }>
|
||||
|
||||
@@ -142,12 +141,12 @@ export class ProjectionListener implements Disposable {
|
||||
const input = this.buffer.createInput<{ session: Session; change: SessionProjectionChange }>(item => [item])
|
||||
const target = sessionId
|
||||
try {
|
||||
this.buffer.own(ctx.sessionProjections.onChanged((session, key, value, seq) => {
|
||||
this.buffer.own(ctx.sessionProjections.onEventChanged((session, changes, seq) => {
|
||||
if (session.id !== target) return
|
||||
this.buffer.capture(() => {
|
||||
input.append({
|
||||
session, change: { type: 'projection', key, value: value as JsonValue, seq } satisfies SessionProjectionChange,
|
||||
})
|
||||
// Every projection definition validates its value before change publication.
|
||||
const change: SessionProjectionChange = { type: 'projection', seq, changes: changes as SessionProjectionChange['changes'] }
|
||||
input.append({ session, change })
|
||||
})
|
||||
}))
|
||||
} catch (error: unknown) {
|
||||
@@ -156,8 +155,8 @@ export class ProjectionListener implements Disposable {
|
||||
}
|
||||
}
|
||||
|
||||
/** Consume sparse per-key projection replacements.
|
||||
* @returns validated replacement frames; projection seqs need not be contiguous. */
|
||||
/** Consume per-event projection changes.
|
||||
* @returns validated change frames, one per committed event; projection seqs need not be contiguous. */
|
||||
async *changes(): AsyncIterable<SessionProjectionChange> {
|
||||
try {
|
||||
for await (const { change } of this.buffer.items()) {
|
||||
|
||||
@@ -553,10 +553,10 @@ export class SessionController extends TypertRemoteService {
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe to sparse Session projection changes.
|
||||
* Subscribe to Session projection changes.
|
||||
* @param request - target Session.
|
||||
* @param signal - subscription cancellation.
|
||||
* @returns readiness and sparse projection changes.
|
||||
* @returns readiness followed by one projection change frame per committed event.
|
||||
*/
|
||||
@Remote({ mode: 'stream' })
|
||||
observeProjections(request: SessionProjectionsRequest, signal: AbortSignal): AsyncIterable<SessionProjectionFrame> {
|
||||
|
||||
@@ -569,12 +569,21 @@ export interface SessionControlBaseline {
|
||||
readonly projections: Readonly<Record<SessionId, SessionProjectionBaseline>>
|
||||
}
|
||||
|
||||
/** One finished projection value and its durable watermark. */
|
||||
/**
|
||||
* Projection changes caused by one committed Session event; the Host emits one
|
||||
* update per event that changed at least one client-visible projection view.
|
||||
*/
|
||||
export interface SessionProjectionUpdate {
|
||||
readonly sessionId: SessionId
|
||||
readonly key: string
|
||||
readonly value: JsonValue
|
||||
/** Seq of the committed event that caused every listed change. */
|
||||
readonly seq: number
|
||||
/**
|
||||
* One entry per client-visible key whose view the event changed, in
|
||||
* registration order without duplicate keys, with the key's new value as
|
||||
* `val`. A key absent from `changes` is unchanged; unlike a baseline, absence
|
||||
* does not mean an absent capability.
|
||||
*/
|
||||
readonly changes: readonly { readonly key: string; readonly val: SessionProjectionValue }[]
|
||||
}
|
||||
|
||||
/** Host-wide live state stream. Each generation starts with exactly one baseline. */
|
||||
@@ -652,10 +661,10 @@ export type SessionEventFrame =
|
||||
| { readonly type: 'ready' }
|
||||
| SessionEventEntry
|
||||
|
||||
/** One sparse per-key replacement; different keys can share a seq. */
|
||||
/** Projection changes caused by one committed event of the observed Session. */
|
||||
export type SessionProjectionChange = { readonly type: 'projection' } & Omit<SessionProjectionUpdate, 'sessionId'>
|
||||
|
||||
/** Subscription readiness without a baseline, followed by per-key changes. */
|
||||
/** Subscription readiness without a baseline, followed by one change frame per committed event. */
|
||||
export type SessionProjectionFrame =
|
||||
| { readonly type: 'ready' }
|
||||
| SessionProjectionChange
|
||||
|
||||
Reference in new issue
Block a user