feat(bigtable): add lazyPool helper for on-demand session pool opening (#20182)

## Summary

- Adds `bigtable/internal/session` with a `lazyPool` primitive that
opens its underlying `Invoker` on first use. Concurrent callers block
until the open completes; failed opens are NOT cached, so a transient
`proto.Marshal` failure cannot strand the caller for the process
lifetime.
- A nil `*lazyPool` or one with a nil `open` closure returns `(nil,
nil)`, letting callers model "no session support, use fallback" (e.g.,
the write side of a read-only materialized view).
- Adds `transport.InvokeResult` — the value type returned by
`Session.Invoke` — so the session package can declare the `Invoker`
interface without pulling in the full session-pool implementation, which
will land in a follow-up.

## Test plan

- [x] `go build ./bigtable/internal/session/...
./bigtable/internal/transport/...`
- [x] `go vet ./bigtable/internal/session/...
./bigtable/internal/transport/...`
- [x] `go test ./bigtable/internal/session/... -run LazyPool -count=1`
(both `TestLazyPool_NilPoolAndNilOpenReturnNilNil` and
`TestLazyPool_FailedOpenNotCached` pass)
- [x] `go test ./bigtable/internal/session/... -count=1 -race`

Author: sushanb

bigtable/internal/session/lazy_pool.go

@@ -0,0 +1,84 @@
+// 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.
+
+package session
+
+import (
+	"context"
+	"sync"
+
+	btransport "cloud.google.com/go/bigtable/internal/transport"
+)
+
+// Invoker is the narrow surface sessionTable needs from a session
+// pool: dispatch a single virtual RPC and surface the full
+// InvokeResult (response, cluster info, server-side Stats, and the
+// local SentAt timestamp). Satisfied by *btransport.SessionPoolImpl;
+// the interface exists so tests can substitute a fake without
+// standing up a real pool.
+type Invoker interface {
+	Invoke(ctx context.Context, desc btransport.VRpcDescriptor, req interface{}) (btransport.InvokeResult, error)
+}
+
+// lazyPool wraps an Invoker (typically *btransport.SessionPoolImpl)
+// that is opened on first use. Callers invoke get(); the first
+// winner runs the open closure (synchronously — dial + handshake
+// happens here) and stores the result; subsequent callers see the
+// stored pool with no work.
+//
+// Failed opens are NOT cached: the next caller retries. This
+// matters because pool creation can fail transiently (proto.Marshal
+// is the only obvious source today, but future descriptor variants
+// may add more). A permanent error-cache would leave the sessionTable
+// stuck for the process lifetime.
+//
+// A nil *lazyPool or one with a nil open closure returns (nil, nil)
+// — "no session support, use fallback." Used for the write side of
+// materialized views (read-only).
+type lazyPool struct {
+	mu   sync.Mutex
+	pool Invoker
+	open func() (Invoker, error)
+}
+
+// get returns the underlying pool, opening it on first call.
+// Concurrent callers block until the open completes.
+func (l *lazyPool) get() (Invoker, error) {
+	if l == nil || l.open == nil {
+		return nil, nil
+	}
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if l.pool != nil {
+		return l.pool, nil
+	}
+	p, err := l.open()
+	if err != nil {
+		return nil, err
+	}
+	l.pool = p
+	return p, nil
+}
+
+// opened reports whether the pool has been opened yet — for tests
+// and for the sessionz debug UI which wants to render "read pool:
+// not yet opened" vs a live pool.
+func (l *lazyPool) opened() bool {
+	if l == nil {
+		return false
+	}
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.pool != nil
+}

bigtable/internal/session/lazy_pool_test.go

@@ -0,0 +1,119 @@
+// 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.
+
+package session
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+
+	btransport "cloud.google.com/go/bigtable/internal/transport"
+)
+
+// stubInvoker is a minimal Invoker for tests that only need identity —
+// distinct from the shared fakeInvoker to avoid coupling this test file
+// to table_test.go's fixture.
+type stubInvoker struct{ tag string }
+
+func (s *stubInvoker) Invoke(_ context.Context, _ btransport.VRpcDescriptor, _ interface{}) (btransport.InvokeResult, error) {
+	return btransport.InvokeResult{}, nil
+}
+
+// TestLazyPool_NilPoolAndNilOpenReturnNilNil verifies the "no session
+// support" contract: a nil *lazyPool or one with a nil open closure
+// returns (nil, nil). Used for the write side of MatView (spec #11).
+func TestLazyPool_NilPoolAndNilOpenReturnNilNil(t *testing.T) {
+	var nilPool *lazyPool
+	if p, err := nilPool.get(); p != nil || err != nil {
+		t.Errorf("nil-receiver get() = (%v, %v), want (nil, nil)", p, err)
+	}
+
+	empty := &lazyPool{}
+	if p, err := empty.get(); p != nil || err != nil {
+		t.Errorf("nil-open get() = (%v, %v), want (nil, nil)", p, err)
+	}
+	if empty.opened() {
+		t.Error("empty.opened() = true, want false")
+	}
+}
+
+// TestLazyPool_FailedOpenNotCached verifies SESSION_SPEC.md #11: failed
+// opens MUST NOT be cached. A transient proto.Marshal failure (or any
+// other opener error) MUST NOT strand the table for the process
+// lifetime. The next get() call re-invokes open().
+//
+// The counterfactual: if we DID cache the failure, calls == 1 after N
+// gets. The invariant is that calls == N.
+func TestLazyPool_FailedOpenNotCached(t *testing.T) {
+	var (
+		mu       sync.Mutex
+		calls    int
+		failNext = true
+		succeed  = &stubInvoker{tag: "opened"}
+		wantErr  = errors.New("marshal boom")
+	)
+	l := &lazyPool{
+		open: func() (Invoker, error) {
+			mu.Lock()
+			defer mu.Unlock()
+			calls++
+			if failNext {
+				return nil, wantErr
+			}
+			return succeed, nil
+		},
+	}
+
+	// Four failing opens — each MUST re-invoke the closure.
+	for i := 0; i < 4; i++ {
+		p, err := l.get()
+		if p != nil {
+			t.Errorf("call #%d: pool = %v, want nil on open failure", i+1, p)
+		}
+		if !errors.Is(err, wantErr) {
+			t.Errorf("call #%d: err = %v, want %v", i+1, err, wantErr)
+		}
+		if l.opened() {
+			t.Errorf("call #%d: opened() = true after failure; failure MUST NOT be cached", i+1)
+		}
+	}
+	mu.Lock()
+	if calls != 4 {
+		t.Errorf("open() invocations after 4 failing gets = %d, want 4 — failed opens were cached, violating spec #11", calls)
+	}
+	// Now flip: next open succeeds. Subsequent gets MUST return the
+	// cached success (open() invoked exactly one more time, not four).
+	failNext = false
+	mu.Unlock()
+
+	for i := 0; i < 4; i++ {
+		p, err := l.get()
+		if err != nil {
+			t.Errorf("post-success call #%d: err = %v, want nil", i+1, err)
+		}
+		if p != succeed {
+			t.Errorf("post-success call #%d: pool = %v, want cached stubInvoker", i+1, p)
+		}
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if calls != 5 {
+		t.Errorf("total open() invocations = %d, want 5 (4 failing + 1 successful cached) — successful opens MUST be cached", calls)
+	}
+	if !l.opened() {
+		t.Error("opened() = false after successful open, want true")
+	}
+}

bigtable/internal/transport/invoke_result.go

@@ -0,0 +1,53 @@
+// 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.
+
+package internal
+
+import (
+	"time"
+
+	spb "cloud.google.com/go/bigtable/apiv2/bigtablepb"
+)
+
+// InvokeResult carries the full set of outputs from a single Invoke call.
+//
+// Fields:
+//   - Response: decoded vRPC payload (typed per VRpcDescriptor.Decode); nil on error.
+//   - ClusterInfo: server-reported routing/cluster identity; may be set on
+//     both success and error paths if the server included it.
+//   - Stats: server-reported per-request statistics (notably BackendLatency);
+//     nil if the server did not populate Stats on the success frame.
+//   - SentAt: local monotonic timestamp captured immediately before the vRPC
+//     frame was handed to the bidi Send. Used downstream to derive
+//     client-side blocking latency (sentAt - attemptStart).
+//
+// ErrorResponse.RetryInfo from the server is plumbed via the returned error
+// using gRPC status details — callers can extract it with
+// status.FromError(err).Details() and type-asserting to *errdetails.RetryInfo.
+type InvokeResult struct {
+	Response    interface{}
+	ClusterInfo *spb.ClusterInformation
+	Stats       *spb.SessionRequestStats
+	// SentAt is a local monotonic timestamp captured immediately before
+	// the vRPC frame is handed to the bidi Send. Used downstream to
+	// derive client-side blocking latency (sentAt - attemptStart).
+	SentAt   time.Time
+	PeerInfo *spb.PeerInfo
+	// RPCIDOnSession is the per-session monotonic id of this call
+	// (1, 2, 3, …). Distinguishes warm-up vRPCs (small id) from
+	// established-session vRPCs.
+	RPCIDOnSession int64
+	// TransportLatency = AttemptLatency - BackendLatency.
+	TransportLatency time.Duration
+}