perf(spanner): synchronous multiplexed session hand-off for non-stream callers

handwritten/spanner/src/common.ts

@@ -18,7 +18,6 @@ import {grpc, CallOptions, Operation as GaxOperation} from 'google-gax';
 import {protos} from '@google-cloud/spanner-api';
 import instanceAdmin = protos.google;
 import databaseAdmin = protos.google;
-import {Spanner} from '.';
 
 export type IOperation = instanceAdmin.longrunning.IOperation;
 
@@ -113,13 +112,13 @@ export function getCommonHeaders(
   const headers: {[k: string]: string} = {};
 
   if (
-    process.env.SPANNER_ENABLE_END_TO_END_TRACING === 'true' ||
+    process.env.SPANNER_ENABLE_END_TO_END_TRACING?.toLowerCase() === 'true' ||
     enableTracing
   ) {
     headers[END_TO_END_TRACING_HEADER] = 'true';
   }
 
-  if (Spanner.isAFEServerTimingEnabled()) {
+  if (process.env.SPANNER_DISABLE_AFE_SERVER_TIMING?.toLowerCase() !== 'true') {
     headers[AFE_SERVER_TIMING_HEADER] = 'true';
   }
 

handwritten/spanner/src/database.ts

@@ -38,7 +38,11 @@
 } from 'google-gax';
 import {Backup} from './backup';
 import {BatchTransaction, TransactionIdentifier} from './batch-transaction';
-import {SessionFactory, SessionFactoryInterface} from './session-factory';
+import {
+  GetSessionCallback,
+  SessionFactory,
+  SessionFactoryInterface,
+} from './session-factory';
 import {protos} from '@google-cloud/spanner-api';
 import google = protos.google;
 import databaseAdmin = protos.google;
@@ -2483,11 +2487,11 @@
     if (gaxOpts) {
       const gax = gaxOpts as GetDatabaseRolesOptions;
       if (gax.pageSize !== undefined) {
         (reqOpts as any).pageSize ??= gax.pageSize;
         delete gax.pageSize;
       }
       if (gax.pageToken !== undefined) {
         (reqOpts as any).pageToken ??= gax.pageToken;
         delete gax.pageToken;
       }
     }
@@ -2531,7 +2535,7 @@
     callback?: PoolRequestCallback,
   ): void | Promise<Session> {
     const sessionFactory_ = this.sessionFactory_;
-    sessionFactory_.getSessionForReadWrite((err, session) => {
+    const onSession: GetSessionCallback = (err, session) => {
       if (err) {
         callback!(err as ServiceError, null);
         return;
@@ -2544,7 +2548,17 @@
         sessionFactory_.release(session!);
         callback!(err, ...args);
       });
-    });
+    };
+
+    const session = sessionFactory_.isMultiplexedEnabledForRW?.()
+      ? sessionFactory_.getSessionSync?.()
+      : null;
+    if (session) {
+      onSession(null, session);
+      return;
+    }
+
+    sessionFactory_.getSessionForReadWrite(onSession);
   }
 
   /**
@@ -3025,16 +3039,30 @@
       callback!(error, rows, stats!, metadata!);
     };
 
-    this.sessionFactory_.getSession((error, session) => {
+    const onSession: GetSessionCallback = (error, session) => {
       if (error) {
         complete(error as grpc.ServiceError);
         return;
       }
 
       streamSpan.addEvent('Using Session', {'session.id': session?.id});
-      snapshot = session!.snapshot(options, this.queryOptions_);
-      this._runOnSnapshot(snapshot, session!, query, complete);
-    });
+      try {
+        snapshot = session!.snapshot(options, this.queryOptions_);
+        this._runOnSnapshot(snapshot, session!, query, complete);
+      } catch (syncError) {
+        // Defer error delivery via nextTick so callback callers never experience
+        // synchronous callback execution (Zalgo) when getSessionSync() returns synchronously.
+        process.nextTick(() => complete(syncError as grpc.ServiceError));
+      }
+    };
+
+    const session = this.sessionFactory_.getSessionSync?.();
+    if (session) {
+      onSession(null, session);
+      return;
+    }
+
+    this.sessionFactory_.getSession(onSession);
   }
 
   /**
@@ -3081,7 +3109,9 @@
         snapshot.run(query, callback as RunCallback);
       }
     } catch (syncError) {
-      callback(syncError as grpc.ServiceError);
+      // Defer error delivery via nextTick so callback callers never experience
+      // synchronous callback execution (Zalgo) when getSessionSync() returns synchronously.
+      process.nextTick(() => callback(syncError as grpc.ServiceError));
     }
   }
   /**

handwritten/spanner/src/multiplexed-session.ts

@@ -51,6 +51,13 @@
    * @param {GetSessionCallback} callback The callback function.
    */
   getSession(callback: GetSessionCallback): void;
+
+  /**
+   * When called returns a cached multiplexed session synchronously if available.
+   *
+   * @name MultiplexedSessionInterface#getSessionSync
+   */
+  getSessionSync(): Session | null;
 }
 
 /**
@@ -195,6 +202,21 @@
     this._refreshHandle.unref();
   }
 
+  /**
+   * Synchronously returns the cached multiplexed session if available,
+   * or null if no session is currently cached.
+   *
+   * @returns {Session|null} The cached multiplexed session or null.
+   */
+  getSessionSync(): Session | null {
+    if (this._multiplexedSession !== null) {
+      const span = getActiveOrNoopSpan();
+      span.addEvent('Cache hit: has usable multiplexed session');
+      return this._multiplexedSession;
+    }
+    return null;
+  }
+
   /**
    * Retrieves a session asynchronously and invokes a callback with the session details.
    * Note: The callback receives `(null, session)`. To prevent unnecessary allocations on
@@ -207,10 +229,8 @@
    *
    */
   getSession(callback: GetSessionCallback): void {
-    if (this._multiplexedSession !== null) {
-      const session = this._multiplexedSession;
-      const span = getActiveOrNoopSpan();
-      span.addEvent('Cache hit: has usable multiplexed session');
+    const session = this.getSessionSync();
+    if (session !== null) {
       // Use process.nextTick to guarantee asynchronous callback execution ("never release Zalgo").
       // This avoids microtask and Promise allocation overhead while preventing race conditions
       // where callers (such as Database.prototype.runStream) need to return their stream and
@@ -223,7 +243,7 @@
 
     this._getSession()
       .then(session => {
         callback(null, session);
         return null;
       }, callback)
       .catch(err => {
@@ -249,13 +269,13 @@
    *
    */
   async _getSession(): Promise<Session | null> {
-    const span = getActiveOrNoopSpan();
     // Check if the multiplexed session is already available
-    if (this._multiplexedSession !== null) {
-      span.addEvent('Cache hit: has usable multiplexed session');
-      return this._multiplexedSession;
+    const cachedSession = this.getSessionSync();
+    if (cachedSession !== null) {
+      return cachedSession;
     }
 
+    const span = getActiveOrNoopSpan();
     span.addEvent('Waiting for a multiplexed session to become available');
 
     // If initialization is ALREADY in progress, join the existing line!

handwritten/spanner/src/session-factory.ts

@@ -58,6 +58,13 @@ export interface SessionFactoryInterface {
    */
   getSession(callback: GetSessionCallback): void;
 
+  /**
+   * When called returns a cached multiplexed session synchronously if available.
+   *
+   * @name SessionFactoryInterface#getSessionSync
+   */
+  getSessionSync(): Session | null;
+
   /**
    * When called returns a session for paritioned dml.
    *
@@ -168,6 +175,18 @@ export class SessionFactory
     }
   }
 
+  /**
+   * Synchronously returns a cached multiplexed session if multiplexed sessions
+   * are enabled and one is available, otherwise null.
+   *
+   * @returns {Session|null} The cached multiplexed session or null.
+   */
+  getSessionSync(): Session | null {
+    return this.isMultiplexed
+      ? (this.multiplexedSession_?.getSessionSync?.() ?? null)
+      : null;
+  }
+
   /**
    * Retrieves a session, either a regular session or a multiplexed session, based on the environment variable configuration.
    *
@@ -176,15 +195,12 @@ export class SessionFactory
    *
    * @param {GetSessionCallback} callback The callback function.
    */
-
   getSession(callback: GetSessionCallback): void {
     const sessionHandler = this.isMultiplexed
       ? this.multiplexedSession_
       : this.pool_;
 
-    sessionHandler!.getSession((err, session, transaction) =>
-      callback(err, session, transaction),
-    );
+    sessionHandler!.getSession(callback);
   }
 
   /**

handwritten/spanner/test/common-headers.ts

@@ -0,0 +1,82 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import * as assert from 'assert';
+import {describe, it, beforeEach, afterEach} from 'mocha';
+import {
+  AFE_SERVER_TIMING_HEADER,
+  CLOUD_RESOURCE_HEADER,
+  END_TO_END_TRACING_HEADER,
+  getCommonHeaders,
+} from '../src/common';
+
+describe('getCommonHeaders', () => {
+  const resource = 'projects/p/instances/i/databases/d';
+  const originalDisableAfe = process.env.SPANNER_DISABLE_AFE_SERVER_TIMING;
+  const originalEnableE2E = process.env.SPANNER_ENABLE_END_TO_END_TRACING;
+
+  beforeEach(() => {
+    delete process.env.SPANNER_DISABLE_AFE_SERVER_TIMING;
+    delete process.env.SPANNER_ENABLE_END_TO_END_TRACING;
+  });
+
+  afterEach(() => {
+    if (originalDisableAfe !== undefined) {
+      process.env.SPANNER_DISABLE_AFE_SERVER_TIMING = originalDisableAfe;
+    } else {
+      delete process.env.SPANNER_DISABLE_AFE_SERVER_TIMING;
+    }
+    if (originalEnableE2E !== undefined) {
+      process.env.SPANNER_ENABLE_END_TO_END_TRACING = originalEnableE2E;
+    } else {
+      delete process.env.SPANNER_ENABLE_END_TO_END_TRACING;
+    }
+  });
+
+  it('should include resource prefix and AFE timing header by default', () => {
+    const headers = getCommonHeaders(resource);
+    assert.strictEqual(headers[CLOUD_RESOURCE_HEADER], resource);
+    assert.strictEqual(headers[AFE_SERVER_TIMING_HEADER], 'true');
+    assert.strictEqual(headers[END_TO_END_TRACING_HEADER], undefined);
+  });
+
+  it('should omit AFE timing header when disabled via case-insensitive env var', () => {
+    process.env.SPANNER_DISABLE_AFE_SERVER_TIMING = 'TRUE';
+    const headers = getCommonHeaders(resource);
+    assert.strictEqual(headers[AFE_SERVER_TIMING_HEADER], undefined);
+  });
+
+  it('should include AFE timing header when env var is false', () => {
+    process.env.SPANNER_DISABLE_AFE_SERVER_TIMING = 'false';
+    const headers = getCommonHeaders(resource);
+    assert.strictEqual(headers[AFE_SERVER_TIMING_HEADER], 'true');
+  });
+
+  it('should include end-to-end tracing header when enabled via case-insensitive env var', () => {
+    process.env.SPANNER_ENABLE_END_TO_END_TRACING = 'True';
+    const headers = getCommonHeaders(resource);
+    assert.strictEqual(headers[END_TO_END_TRACING_HEADER], 'true');
+  });
+
+  it('should include end-to-end tracing header when enableTracing option is true', () => {
+    const headers = getCommonHeaders(resource, true);
+    assert.strictEqual(headers[END_TO_END_TRACING_HEADER], 'true');
+  });
+
+  it('should omit end-to-end tracing header when env var is false and enableTracing option is false', () => {
+    process.env.SPANNER_ENABLE_END_TO_END_TRACING = 'FALSE';
+    const headers = getCommonHeaders(resource, false);
+    assert.strictEqual(headers[END_TO_END_TRACING_HEADER], undefined);
+  });
+});