chore(spanner): add a channel pool prototype

[olavloite]

Adds a gRPC channel pool specifically tailored for Spanner.

[gemini-code-assist]

Code Review

This pull request introduces a custom gRPC channel pool implementation for Spanner, featuring both static and dynamic load-based channel pools, transaction affinity management, and a call invocation transformer to route RPCs. The feedback highlights several critical issues and optimization opportunities: a resource leak in ChannelPoolChannelAdapter.createCall where acquired leases are never released; potential runtime crashes if the channel pools are configured with zero channels; incorrect transaction load tracking during fallback scenarios in DynamicChannelPool.acquire; and O(N) linear scan performance bottlenecks in MetricsTracerFactory that can be optimized to O(1) lookups. Additionally, suggestions are provided to safely track channel pool ownership to prevent closing user-provided pools, improve timer safety in channel priming, and make StaticChannelPool more consistent by storing the target address.

[gemini-code-assist]

In createCall, the leased channel is acquired but the lease is never released. This will leak the in-flight RPC count, skewing load balancing metrics and preventing the dynamic pool from scaling down. Wrap the returned call in a grpc.InterceptingCall to release the lease when the call completes or is cancelled. Avoid wrapping default pass-through methods in arrow functions to prevent unnecessary closure allocations.

  createCall(
    method: string,
    deadline: any,
    host: any,
    parentCall: any,
    propagateFlags: any,
  ): any {
    // Fallback if call is initiated outside of callInvocationTransformer
    const lease = this.pool.acquire();
    const call = (lease.entry.channel as any).createCall(
      method,
      deadline,
      host,
      parentCall,
      propagateFlags,
    );
    let released = false;
    const releaseOnce = () => {
      if (!released) {
        released = true;
        lease.release();
      }
    };
    return new grpc.InterceptingCall(call, {
      start: (metadata, listener, next) => {
        next(metadata, {
          onReceiveStatus: (status, next) => {
            releaseOnce();
            next(status);
          },
        });
      },
      cancel: next => {
        releaseOnce();
        next();
      },
    });
  }

References

1. Avoid wrapping methods in arrow functions for default cases to prevent unnecessary closure allocations and extra call stack frames.

[gemini-code-assist]

If options.minChannels is configured as 0, this.minChannels will be 0. This allows the pool to scale down to 0 active channels, which will cause all subsequent acquire() calls to throw an error and prevent the pool from ever scaling back up. Enforce a minimum of 1 channel.

Suggested change

	this.minChannels = initialCount;
	this.minChannels = Math.max(1, initialCount);

[gemini-code-assist]

In acquire(), if a Read/Write transaction's pinned channel is closed or drained, the pool falls back to a new active channel, but it fails to increment activeRwTransactions on the new channel. This will cause incorrect load tracking and potentially negative transaction counts when the transaction is reset. Simplify the logic to correctly handle fallback and increment the transaction count.

  acquire(affinity?: TransactionAffinity): ChannelLease {
    let entry: ChannelEntry;

    if (
      affinity?.pinnedEntry &&
      (affinity.pinnedEntry.state === 'ACTIVE' ||
        (affinity.pinnedEntry.state === 'DRAINING' &&
          affinity.kind === AffinityKind.ReadWrite))
    ) {
      entry = affinity.pinnedEntry;
    } else {
      entry = selectPowerOfTwo(this.activeEntries);
      if (affinity) {
        affinity.pinnedEntry = entry;
        if (affinity.kind === AffinityKind.ReadWrite) {
          entry.activeRwTransactions++;
        }
      }
    }

    entry.inFlightRpcs++;
    entry.lastActivity = Date.now();

    // 3. Event-driven scale-up check
    if (entry.inFlightRpcs > this.maxRpcPerChannel) {
      this.maybeScaleUp();
    }

    let released = false;
    return {
      entry,
      release: () => {
        if (!released) {
          released = true;
          entry.inFlightRpcs = Math.max(0, entry.inFlightRpcs - 1);
          entry.lastActivity = Date.now();
          if (entry.state === 'DRAINING') {
            this.checkDrainedEntry(entry);
          }
        }
      },
    };
  }

[gemini-code-assist]

Add a private _ownsChannelPool property to track whether the Spanner instance created the channel pool internally or if it was provided by the user. This prevents closing a shared user-provided channel pool when a single Spanner instance is closed.

Suggested change

	private channelPool_?: ChannelPool;
	private channelPool_?: ChannelPool;
	private _ownsChannelPool = false;

[gemini-code-assist]

Set this._ownsChannelPool = true when creating the channel pool internally.

Suggested change

	if (!channelPoolInstance) {
	channelPoolInstance = createChannelPool(
	address,
	credentials,
	channelOptions,
	poolConfig,
	);
	this.channelPool_ = channelPoolInstance;
	}
	if (!channelPoolInstance) {
	channelPoolInstance = createChannelPool(
	address,
	credentials,
	channelOptions,
	poolConfig,
	);
	this.channelPool_ = channelPoolInstance;
	this._ownsChannelPool = true;
	}

[gemini-code-assist]

Optimize clearCurrentTracer to use O(1) lookup by reconstructing the original key with channel ID '1'.

    if (
      !this._currentOperationTracers.has(operationRequest) &&
      operationRequest
    ) {
      const parts = operationRequest.split('.');
      if (parts.length === 5) {
        parts[3] = '1';
        const originalKey = parts.join('.');
        if (this._currentOperationTracers.has(originalKey)) {
          operationRequest = originalKey;
        }
      }
    }

[gemini-code-assist]

Refactor primeChannel to declare and assign the timer inside the try block and avoid using the non-null assertion operator !. This is safer and prevents potential timer leaks if synchronous errors occur.

Suggested change

	private async primeChannel(entry: ChannelEntry): Promise<void> {
	if (!this.primeSessionName || !this.primeFn) {
	return;
	}
	const primePromise = this.primeFn(entry.channel, this.primeSessionName);
	let timer: NodeJS.Timeout;
	const timeoutPromise = new Promise<never>((_resolve, reject) => {
	timer = setTimeout(() => {
	reject(new Error('Channel priming timed out.'));
	}, this.primeTimeoutMs);
	timer.unref();
	});
	try {
	await Promise.race([primePromise, timeoutPromise]);
	} finally {
	clearTimeout(timer!);
	}
	}
	private async primeChannel(entry: ChannelEntry): Promise<void> {
	if (!this.primeSessionName || !this.primeFn) {
	return;
	}
	let timer: NodeJS.Timeout | undefined;
	try {
	const primePromise = this.primeFn(entry.channel, this.primeSessionName);
	const timeoutPromise = new Promise<never>((_resolve, reject) => {
	timer = setTimeout(() => {
	reject(new Error('Channel priming timed out.'));
	}, this.primeTimeoutMs);
	timer.unref();
	});
	await Promise.race([primePromise, timeoutPromise]);
	} finally {
	if (timer) {
	clearTimeout(timer);
	}
	}
	}

[gemini-code-assist]

Add a private target property to StaticChannelPool to store the target address, making it consistent with DynamicChannelPool and avoiding direct dependency on the entries array in getTarget().

export class StaticChannelPool implements ChannelPool {
  private readonly entries: ChannelEntry[];
  private readonly target: string;

  constructor(
    target: string,
    credentials: grpc.ChannelCredentials,
    channelOptions: Record<string, any>,
    options?: StaticChannelPoolOptions,
  ) {
    this.target = target;

[gemini-code-assist]

Enforce a minimum of 1 channel in StaticChannelPool to prevent runtime errors if numChannels is configured as 0.

Suggested change

	let count = options?.numChannels ?? 4;
	let count = Math.max(1, options?.numChannels ?? 4);

[gemini-code-assist]

Return the stored target property directly instead of accessing the first entry's channel.

Suggested change

	getTarget(): string {
	return this.entries.length > 0 ? this.entries[0].channel.getTarget() : '';
	}
	getTarget(): string {
	return this.target;
	}