Use the runtime-provided ingest transport when available

.changeset/tidy-donuts-brush.md

@@ -0,0 +1,5 @@
+---
+'@vercel/flags-core': patch
+---
+
+Use the runtime-provided ingest transport when available

packages/vercel-flags-core/src/utils/ingest.ts

@@ -3,6 +3,7 @@ import { version } from '../../package.json';
 import type { Auth } from '../controller/auth';
 import type { MetricEnvironment } from '../types';
 import { getRetryDelayMs } from './backoff';
+import { getRuntimeIngest } from './runtime-ingest';
 import type { FlushReason } from './scheduler';
 import type { IngestEvent, UsageEvent } from './usage/events';
 
@@ -70,13 +71,49 @@ async function getIngestHeaders(
   };
 }
 
+/**
+ * Headers for the runtime-provided ingest transport. The runtime attributes
+ * the caller itself, so OIDC tokens are omitted; only an SDK key is included
+ * when configured.
+ */
+function getRuntimeIngestHeaders(
+  options: IngestOptions,
+  flushReason: FlushReason,
+): Record<string, string> {
+  return {
+    'Content-Type': 'application/json',
+    ...(options.auth.sdkKey
+      ? { Authorization: `Bearer ${options.auth.sdkKey}` }
+      : null),
+    'User-Agent': `VercelFlagsCore/${version}`,
+    [FLUSH_REASON_HEADER]: flushReason,
+    ...((options.metricEnvironment ?? process.env.VERCEL_ENV)
+      ? {
+          'X-Vercel-Env':
+            options.metricEnvironment ?? (process.env.VERCEL_ENV as string),
+        }
+      : null),
+    ...(isDebugMode ? { 'x-vercel-debug-ingest': '1' } : null),
+  };
+}
+
 export async function sendIngestEvents(
   options: IngestOptions,
   events: UsageEvent[],
   flushId: number,
   flushReason: FlushReason,
 ): Promise<void> {
-  const eventsToSend = events.map((event) => event.ingestEvent());
+  let eventsToSend = events.map((event) => event.ingestEvent());
+
+  const runtimeIngest = getRuntimeIngest();
+  if (runtimeIngest) {
+    const headers = getRuntimeIngestHeaders(options, flushReason);
+    // Events the runtime does not accept fall through to the HTTP transport.
+    eventsToSend = eventsToSend.filter(
+      (event) => !runtimeIngest({ headers, body: [event] }),
+    );
+    if (eventsToSend.length === 0) return;
+  }
 
   for (let i = 0; i < eventsToSend.length; i += MAX_EVENTS_PER_REQUEST) {
     await sendIngestChunk(

packages/vercel-flags-core/src/utils/runtime-ingest.ts

@@ -0,0 +1,27 @@
+import type { IngestEvent } from './usage/events';
+
+export type RuntimeIngest = (payload: {
+  headers: Record<string, string>;
+  body: IngestEvent[];
+}) => boolean;
+
+const FLAGS_CONTEXT_SYMBOL = Symbol.for('@vercel/flags-context');
+
+/**
+ * Returns the ingest transport provided by the runtime, if available.
+ */
+export function getRuntimeIngest(): RuntimeIngest | undefined {
+  try {
+    const context = (
+      globalThis as typeof globalThis & {
+        [key: symbol]: { ingest?: unknown } | undefined;
+      }
+    )[FLAGS_CONTEXT_SYMBOL];
+
+    return typeof context?.ingest === 'function'
+      ? (context.ingest as RuntimeIngest)
+      : undefined;
+  } catch {
+    return undefined;
+  }
+}

packages/vercel-flags-core/src/utils/scheduler.ts

@@ -6,7 +6,11 @@ const IDLE_FLUSH_WAIT_MS = 5000;
 const IDLE_FLUSH_JITTER_RATIO = 0.2;
 const MAX_FLUSH_WAIT_MS = 60000;
 
-export type FlushReason = 'idle_timeout' | 'max_timeout' | 'shutdown';
+export type FlushReason =
+  | 'idle_timeout'
+  | 'max_timeout'
+  | 'shutdown'
+  | 'immediate';
 
 /**
  * Schedule helper that flushes when any of the following occur:

packages/vercel-flags-core/src/utils/usage-tracker.test.ts

@@ -1227,3 +1227,176 @@ describe('UsageTracker', () => {
     });
   });
 });
+
+describe('runtime ingest transport', () => {
+  const FLAGS_CONTEXT_SYMBOL = Symbol.for('@vercel/flags-context');
+
+  type RuntimeIngestPayload = {
+    headers: Record<string, string>;
+    body: { type: string; ts: number; payload: object }[];
+  };
+
+  let ingestMock: ReturnType<
+    typeof vi.fn<(p: RuntimeIngestPayload) => boolean>
+  >;
+
+  beforeEach(() => {
+    ingestMock = vi.fn<(p: RuntimeIngestPayload) => boolean>();
+    Object.defineProperty(globalThis, FLAGS_CONTEXT_SYMBOL, {
+      value: { ingest: ingestMock },
+      configurable: true,
+    });
+  });
+
+  afterEach(() => {
+    delete (globalThis as Record<symbol, unknown>)[FLAGS_CONTEXT_SYMBOL];
+  });
+
+  it('delivers events through the runtime without fetch', async () => {
+    ingestMock.mockReturnValue(true);
+
+    const tracker = createTracker();
+    tracker.trackEvaluation({
+      flagKey: 'my-flag',
+      variant: 'on',
+      reason: ResolutionReason.RULE_MATCH,
+    });
+
+    await vi.waitFor(() => expect(ingestMock).toHaveBeenCalledTimes(1));
+
+    const { headers, body } = ingestMock.mock.calls[0]![0];
+    expect(headers.Authorization).toBe('Bearer test-key');
+    expect(headers[FLUSH_REASON_HEADER]).toBe('immediate');
+    expect(headers[EVALUATING_OIDC_TOKEN_HEADER]).toBeUndefined();
+    expect(body).toHaveLength(1);
+    expect(body[0]!.type).toBe('FLAG_EVALUATION');
+
+    // The flush is registered with waitUntil, but it is already settled when
+    // the runtime accepted every event, so it never extends the invocation.
+    expect(waitUntilMock).toHaveBeenCalledTimes(1);
+    await waitUntilMock.mock.calls[0]![0];
+
+    expect(fetchMock).not.toHaveBeenCalled();
+    expect(getVercelOidcTokenMock).not.toHaveBeenCalled();
+  });
+
+  it('omits the authorization header without an SDK key', async () => {
+    ingestMock.mockReturnValue(true);
+
+    const resolveToken = vi.fn();
+    const tracker = new UsageTracker({
+      waitUntil,
+      auth: {
+        sdkKey: undefined,
+        resolveToken,
+        resolveBundledDefinitionsLookup: () =>
+          Promise.resolve({ type: 'project-id' as const, projectId: 'prj_1' }),
+      },
+      host: 'https://example.com',
+      fetch: fetchMock,
+    });
+    tracker.trackEvaluation({
+      flagKey: 'my-flag',
+      variant: 'on',
+      reason: ResolutionReason.RULE_MATCH,
+    });
+
+    await vi.waitFor(() => expect(ingestMock).toHaveBeenCalledTimes(1));
+
+    const { headers } = ingestMock.mock.calls[0]![0];
+    expect(headers.Authorization).toBeUndefined();
+    expect(headers[EVALUATING_OIDC_TOKEN_HEADER]).toBeUndefined();
+    expect(resolveToken).not.toHaveBeenCalled();
+    expect(getVercelOidcTokenMock).not.toHaveBeenCalled();
+    expect(fetchMock).not.toHaveBeenCalled();
+  });
+
+  it('delivers each event separately', async () => {
+    ingestMock.mockReturnValue(true);
+
+    const tracker = createTracker();
+    tracker.trackRead();
+    tracker.trackEvaluation({
+      flagKey: 'my-flag',
+      variant: 'on',
+      reason: ResolutionReason.RULE_MATCH,
+    });
+
+    await vi.waitFor(() => expect(ingestMock).toHaveBeenCalledTimes(2));
+    const types = ingestMock.mock.calls.flatMap((call) =>
+      call[0].body.map((event) => event.type),
+    );
+    expect(types).toEqual(
+      expect.arrayContaining(['FLAGS_CONFIG_READ', 'FLAG_EVALUATION']),
+    );
+    expect(fetchMock).not.toHaveBeenCalled();
+  });
+
+  it('falls back to fetch for events the runtime does not accept', async () => {
+    ingestMock.mockReturnValue(false);
+    fetchMock.mockImplementation(() => jsonResponse({ ok: true }));
+
+    const tracker = createTracker();
+    tracker.trackEvaluation({
+      flagKey: 'my-flag',
+      variant: 'on',
+      reason: ResolutionReason.RULE_MATCH,
+    });
+
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1));
+    const events = getBody() as SerializedEvaluationEvent[];
+    expect(events).toHaveLength(1);
+    expect(events[0]!.type).toBe('FLAG_EVALUATION');
+  });
+
+  it('drains the HTTP fallback before shutdown resolves', async () => {
+    ingestMock.mockReturnValue(false);
+
+    let resolveFetch!: (response: Response | PromiseLike<Response>) => void;
+    fetchMock.mockImplementation(
+      () => new Promise<Response>((res) => (resolveFetch = res)),
+    );
+
+    const tracker = createTracker();
+    tracker.trackEvaluation({
+      flagKey: 'my-flag',
+      variant: 'on',
+      reason: ResolutionReason.RULE_MATCH,
+    });
+
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1));
+
+    let shutdownResolved = false;
+    const shutdown = tracker.shutdown().then(() => {
+      shutdownResolved = true;
+    });
+
+    // Let any settled promises run: shutdown must still be waiting on the
+    // in-flight fallback send.
+    await new Promise((res) => setTimeout(res, 0));
+    expect(shutdownResolved).toBe(false);
+
+    resolveFetch(jsonResponse({ ok: true }));
+    await shutdown;
+
+    const events = getBody() as SerializedEvaluationEvent[];
+    expect(events).toHaveLength(1);
+    expect(events[0]!.type).toBe('FLAG_EVALUATION');
+  });
+
+  it('uses the scheduler when the runtime does not provide a transport', async () => {
+    delete (globalThis as Record<symbol, unknown>)[FLAGS_CONTEXT_SYMBOL];
+    fetchMock.mockImplementation(() => jsonResponse({ ok: true }));
+
+    const tracker = createTracker();
+    tracker.trackEvaluation({
+      flagKey: 'my-flag',
+      variant: 'on',
+      reason: ResolutionReason.RULE_MATCH,
+    });
+
+    expect(waitUntilMock).toHaveBeenCalledTimes(1);
+    await tracker.shutdown();
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+});

packages/vercel-flags-core/src/utils/usage-tracker.ts

@@ -1,6 +1,7 @@
 import type { WaitUntil } from '../types';
 import { type IngestOptions, sendIngestEvents } from './ingest';
 import { getRequestContext } from './request-context';
+import { getRuntimeIngest } from './runtime-ingest';
 import { type FlushReason, Scheduler } from './scheduler';
 import {
   FlagsConfigReadEvent,
@@ -21,6 +22,8 @@ export class UsageTracker {
 
   private options: IngestOptions;
   private scheduler: Scheduler;
+  private waitUntil: WaitUntil;
+  private inflightFlushes = new Set<Promise<void>>();
 
   private trackedRequests = new WeakSet<object>();
 
@@ -29,6 +32,7 @@ export class UsageTracker {
 
   constructor(options: IngestOptions & { waitUntil: WaitUntil }) {
     this.options = options;
+    this.waitUntil = options.waitUntil;
     this.scheduler = new Scheduler(
       (reason) => this.flushEvents(reason),
       options.waitUntil,
@@ -46,6 +50,11 @@ export class UsageTracker {
     // Safety net for events tracked after the drained batch reset; if the
     // drained flush already sent everything this returns early (maps cleared).
     await this.flushEvents('shutdown');
+
+    // Drain immediate flushes whose events fell back to the HTTP transport;
+    // their events left the maps before the async send started, so the flush
+    // above cannot cover them.
+    await Promise.all([...this.inflightFlushes]);
   }
 
   /**
@@ -64,7 +73,7 @@ export class UsageTracker {
 
       this.readEvents.push(new FlagsConfigReadEvent(headers, options));
 
-      this.scheduler.scheduleFlush();
+      this.requestFlush();
     } catch (error) {
       // trackRead should never throw, but log the error
       console.error('@vercel/flags-core: Failed to record event:', error);
@@ -94,7 +103,7 @@ export class UsageTracker {
       }
 
       // always schedule to reset the timer
-      this.scheduler.scheduleFlush();
+      this.requestFlush();
     } catch (error) {
       console.error(
         '@vercel/flags-core: Failed to record evaluation event:',
@@ -103,6 +112,31 @@ export class UsageTracker {
     }
   }
 
+  /**
+   * Flushes immediately when the runtime provides an ingest transport,
+   * otherwise falls back to the time-based scheduler.
+   */
+  private requestFlush(): void {
+    if (getRuntimeIngest()) {
+      // Track the flush so shutdown() can drain it: events the runtime does
+      // not accept fall back to the async HTTP transport, which outlives this
+      // synchronous call. When the runtime accepts everything the promise is
+      // already settled, so neither waitUntil nor shutdown() waits on it.
+      const flush = this.flushEvents('immediate').catch((error) => {
+        console.error('@vercel/flags-core: Failed to flush events:', error);
+      });
+      this.inflightFlushes.add(flush);
+      void flush.finally(() => this.inflightFlushes.delete(flush));
+      try {
+        this.waitUntil?.(flush);
+      } catch {
+        // waitUntil is best-effort; shutdown() still drains the flush.
+      }
+    } else {
+      this.scheduler.scheduleFlush();
+    }
+  }
+
   /**
    * Send all events to the ingest service
    */