Streaming & Polling API Design Guidelines in KHI

This skill provides guidelines and implementation patterns for creating Server-Side Streaming RPCs with transparent polling fallbacks in KHI (supporting non-streaming hosting environments like Google App Engine Standard).

---

1. High-Level Architecture Overview

To support environments where Server-Side Streaming is unavailable, KHI defines paired RPCs in Protobuf and uses a client-side Connect Interceptor to map streaming calls to polling Unary calls transparently.

flowchart LR
    subgraph Frontend
        Client[Caller / UI Code]
        Interceptor[Connect Interceptor]
        Client -->|for await res of ...| Interceptor
    end

    subgraph Backend
        StreamHandler[Streaming RPC Handler]
        PollHandler[Polling Unary RPC Handler]
        JobMgr[AsyncJobManager / StateObserver]
        PollHandler --> JobMgr
        StreamHandler --> JobMgr
    end

    Interceptor -->|"Direct Streaming (Default)"| StreamHandler
    Interceptor -->|"Unary Polling (when ?pollLegacy=true)"| PollHandler

---

2. API Classification & Naming Conventions

Streaming RPCs in KHI fall into two distinct categories:

A. State Observation (Watch-style)

Monitors an ongoing state or resource without triggering background computation.

• Streaming RPC: Watch<Resource> (e.g., WatchPopup, WatchServerStat, WatchInspections, WatchIndexProgress).
• Polling RPC: Pull<Resource> (e.g., PullPopup, PullServerStat, PullInspections, PullIndexProgress).
• Semantics: Returns the current snapshot of the resource state.

B. Heavy Processing / Job Execution (Job-style)

Initiates a long-running, CPU/IO-heavy calculation or session-loading task.

• Streaming RPC: <Action> (e.g., OpenWorkbench, FilterTimeline).
• Polling RPC: <Action>Sync (e.g., OpenWorkbenchSync, FilterTimelineSync).
• Cancellation RPC: Cancel<Action>Sync (e.g., CancelOpenWorkbenchSync, CancelFilterTimelineSync).
• Request Parameters:
  • Initial poll: job_id = "" plus request payload (parameters, queries).
  • Subsequent polls: job_id = "<returned_id>".
• Response Parameters:
  • job_id: Identifier for subsequent poll requests.
  • is_done: Boolean indicating task completion.
  • progress: Optional intermediate progress message.
  • result: Optional final result payload (populated when is_done = true).
  • error_message: Populated if the background job failed.

---

3. Backend Implementation Patterns

Utility packages for state observation and async jobs reside in pkg/server/streamingutil/.

A. Snapshot Observation (streamingutil.StateObserver)

Use StateObserver[T] when subscribers need to be notified of atomic state changes:

observer := streamingutil.NewStateObserver[MyState](initialState)

// Update state
observer.SetState(newState)

// Poll current state
current := observer.GetState()

// Subscribe for changes (used in streaming RPC)
ch, unsubscribe := observer.Subscribe()
defer unsubscribe()

B. Long-Running Job Execution (streamingutil.AsyncJobManager)

Use AsyncJobManager when managing background tasks:

mgr := streamingutil.NewAsyncJobManager()
defer mgr.Close() // Cancels all running jobs and evicts finished jobs

Key Lifecycle Management Rules

1. Abandonment Timeout:
   • If a client abruptly stops polling (e.g., network drop, tab closed without cancellation), the job's context is canceled after a timeout (default 15 seconds) to prevent CPU resource leaks.
2. TTL Eviction:
   • After a job reaches completion or error, its cached result is evicted via ttlcleaner.TTLCleaner (default 1 minute) to free memory.
3. Superceded Job Cancellation:
   • When a new request arrives for the same manager/session with job_id == "", any currently active job in that manager is immediately canceled.
4. Independent Job Context:
   • In polling mode, HTTP request contexts are short-lived. Always run background jobs under a detached context (context.WithCancel(context.Background())), tied to the job's cancellation lifecycle.

---

4. Frontend Connect Interceptor Pattern

Located in web/src/app/services/api/legacy-polling.interceptor.ts.

A. Transparent Streaming Adapter

The interceptor detects req.stream === true and wraps polling unary calls into an AsyncIterable<Message>:

return {
  stream: true,
  service: req.service,
  method: req.method,
  header: new Headers(),
  trailer: new Headers(),
  message: adapter(req, unaryTransport),
};

This guarantees zero code changes at the client call-sites (for await (const res of client.openWorkbench(...))).

B. Extracting the Initial Request Message

[!WARNING] Do NOT use for await (const msg of req.message) { return msg; }. Returning early from for await triggers the iterator return() protocol. In Angular/esbuild environments, Connect wraps unary inputs in [input], whose iterator does not implement .return(), resulting in TypeError: obj[k] is not a function.

Always use direct iterator access:

const iterator = iterable[Symbol.asyncIterator]();
const res = await iterator.next();
if (res.done || res.value === undefined) {
  throw new Error("No request message received.");
}
return res.value;

C. Automatic Cancellation on Client Abort

Inside async generator adapters, always wire up cancellation in a finally block when req.signal.aborted is true:

try {
  while (!req.signal.aborted) {
    const res = await wbClient.filterTimelineSync(...);
    yield ...;
    if (res.isDone) return;
    await delay(300, req.signal);
  }
} finally {
  if (req.signal.aborted && jobId) {
    void wbClient.cancelFilterTimelineSync({ workbenchId, jobId });
  }
}

D. Mode Toggling

Legacy polling mode is activated via:

• URL query parameter: ?pollLegacy=true (for testing and debugging).
• Environment flag: environment.usePollingLegacy (in environment.ts / environment.prod.ts).