8000
Skip to content

fix(bigtable): sessionTableHandle self-heals across cache eviction #20296

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sushanb merged 4 commits into from
Aug 3, 2026
Merged

fix(bigtable): sessionTableHandle self-heals across cache eviction #20296

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
82cd0a3
fix(bigtable): sessionTableHandle self-heals across cache eviction (b…
sushanb Aug 1, 2026
2d64572
fixup(bigtable): sessionTableHandle round-2 review pass
sushanb Aug 3, 2026
5e66829
fixup(bigtable): sessionTableHandle round-2 review nits
sushanb Aug 3, 2026
2552f50
fixup(bigtable): cache resolved successor + path-compress dispatch
sushanb Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions bigtable/docs/specs/SESSION_POOL_SPEC.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@ Any new pool method that reads `p.picker.Name()` from a method called by `Checko
**`TableShim` (`bigtable/table_shim.go`) — mechanism layer.**
- Implements the public `TableAPI` — this is what user code holds when running mixed-mode. Owns `(classic TableAPI, session SessionTableApi, diverter *Diverter)`. Any of `session` / `diverter` may be nil → **shim degrades to classic-only** silently. This is the fallback contract when session support is not enabled or the pool failed to open.
- Per-call routing rule: `if !t.useSession() { classic } else { session }`. `useSession()` = `session != nil && diverter != nil && diverter.UseSession()`.
- **`session` (typed `session.TableAPI`) is backed by a concrete `*sessionTableHandle` that self-heals across cache eviction.** The pointer TableShim caches at Open time stays valid for the shim's lifetime; the wrapper routes evicted RPCs through `cache.getOrOpen`. Mechanism lives in `session_table_cache.go` (see `sessionTableHandle.dispatch` + `resolveSuccessor`); guarded by `TestSessionTableHandle_EvictedSelfHeals` / `TestSessionTableHandle_SweeperEvictionSelfHeals`.
- Owns **all proto ↔ `bigtable.Row` conversion** at the boundary — the `internal/session` package stays proto-native (never sees `bigtable.Row`, `Mutation`, `Filter`, etc.). This is how the two data planes stay decoupled.
- Method routability is **fixed by shape**, not by config — the shim MUST NOT attempt to route an operation whose vRPC equivalent doesn't exist. Enforced today as:

Expand Down
274 changes: 0 additions & 274 deletions bigtable/open_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ package bigtable

import (
"context"
"sync/atomic"
"testing"
"time"

Expand Down Expand Up @@ -628,276 +627,3 @@ func TestGetOrCreateSession_NilSessionImplReturnsNil(t *testing.T) {
t.Errorf("getOrCreateSessionMaterializedView(nil sessionImpl) = %v, want nil", got)
}
}

// ─── sessionTableCache tests ──────────────────────────────────────────

// newTestSessionTableCache builds a cache with an injectable clock.
// Sweep interval is intentionally long — tests drive sweepOnce
// directly instead of waiting on the background ticker, which is
// wall-clock and flaky under CI scheduler pressure (#20266). TTL is
// set per test.
func newTestSessionTableCache(t *testing.T, ttl time.Duration, clock *fakeClock) *sessionTableCache {
t.Helper()
c := newSessionTableCache(ttl, 1*time.Hour, clock.now)
t.Cleanup(c.close)
return c
}

// openNoop returns an openFn that constructs a *noopSessionTable
// stamped with the given key. Used by cache-internal tests that
// want to inspect which key the cache asked to open.
func openNoop(key string) func() session.TableAPI {
return func() session.TableAPI { return &noopSessionTable{key: key} }
}

// openClosing returns an openFn that constructs a
// *closeCountingTable stamped with the given key, atomically
// incrementing counter on every Close call. Counter is atomic so
// sweeper-goroutine writes don't race the test's Load.
func openClosing(key string, counter *atomic.Int32) func() session.TableAPI {
return func() session.TableAPI {
return &closeCountingTable{noopSessionTable{key: key}, counter}
}
}

// fakeClock is a monotonic clock that only advances on explicit
// advance() calls. Concurrency: single writer via advance(), many
// readers via now() — protected by an atomic.
type fakeClock struct{ nano atomic.Int64 }

func newFakeClock(start time.Time) *fakeClock {
c := &fakeClock{}
c.nano.Store(start.UnixNano())
return c
}
func (c *fakeClock) now() time.Time { return time.Unix(0, c.nano.Load()) }
func (c *fakeClock) advance(d time.Duration) { c.nano.Add(int64(d)) }

// TestSessionTableCache_HandleIsCacheEntry pins that the returned
// handle satisfies session.TableAPI, wraps the underlying api, and
// is the same value the cache holds (identity check on repeat Open).
func TestSessionTableCache_HandleIsCacheEntry(t *testing.T) {
clock := newFakeClock(time.Unix(1_700_000_000, 0))
c := newTestSessionTableCache(t, 1*time.Hour, clock)

h1 := c.getOrOpen("tbl:t", openNoop("tbl:t")).(*sessionTableHandle)
h2 := c.getOrOpen("tbl:t", openNoop("tbl:t")).(*sessionTableHandle)
if h1 != h2 {
t.Errorf("repeat getOrOpen on same key = distinct handles: h1=%p h2=%p", h1, h2)
}
if _, ok := h1.api.(*noopSessionTable); !ok {
t.Errorf("handle.api = %T, want *noopSessionTable", h1.api)
}
}

// TestSessionTableCache_ReadRowTouchesLastAccess pins that the
// wrapper's ReadRow updates lastAccess so a caller polling every
// ReadRow keeps the entry alive.
func TestSessionTableCache_ReadRowTouchesLastAccess(t *testing.T) {
clock := newFakeClock(time.Unix(1_700_000_000, 0))
c := newTestSessionTableCache(t, 1*time.Hour, clock)

h := c.getOrOpen("tbl:t", openNoop("tbl:t")).(*sessionTableHandle)
before := h.lastAccessNano.Load()

clock.advance(30 * time.Minute)
_, _ = h.ReadRow(context.Background(), &btpb.SessionReadRowRequest{Key: []byte("r")})

after := h.lastAccessNano.Load()
if after <= before {
t.Errorf("ReadRow did not bump lastAccess: before=%d after=%d", before, after)
}
}

// TestSessionTableCache_CloseEvictsAndFires pins that handle.Close
// removes the entry from the cache map AND calls the underlying
// api.Close, and that a subsequent getOrOpen mints a fresh handle
// (the closed one is not resurrected).
func TestSessionTableCache_CloseEvictsAndFires(t *testing.T) {
clock := newFakeClock(time.Unix(1_700_000_000, 0))
var closeCount atomic.Int32
c := newSessionTableCache(1*time.Hour, 1*time.Second, clock.now)
t.Cleanup(c.close)

h1 := c.getOrOpen("tbl:t", openClosing("tbl:t", &closeCount)).(*sessionTableHandle)
if err := h1.Close(); err != nil {
t.Fatalf("h1.Close: %v", err)
}
if got := closeCount.Load(); got != 1 {
t.Errorf("underlying Close called %d times, want 1", got)
}
// Map should no longer contain the key.
c.mu.Lock()
_, still := c.entries["tbl:t"]
c.mu.Unlock()
if still {
t.Error("entry still present after handle.Close()")
}
// Second Open mints a fresh handle.
h2 := c.getOrOpen("tbl:t", openClosing("tbl:t", &closeCount)).(*sessionTableHandle)
if h2 == h1 {
t.Error("getOrOpen after Close returned the evicted handle")
}
// Double-Close on h1 is fully idempotent — closeOnce guards both
// the map removal AND the underlying api.Close call, so the
// counter stays at 1 even after a second h1.Close().
if err := h1.Close(); err != nil {
t.Errorf("h1.Close (idempotent) err = %v, want nil", err)
}
if got := closeCount.Load(); got != 1 {
t.Errorf("underlying Close called %d times after double-Close, want 1 (Close is fully idempotent)", got)
}
}

// TestSessionTableCache_TTLSweepEvictsIdle pins that a sweep evicts
// entries whose lastAccess is older than TTL, and calls the
// underlying Close on eviction.
//
// Drives sweepOnce directly instead of polling on the background
// ticker: the ticker fires at a real-wall-clock cadence and, under CI
// scheduler pressure, may not run within the assertion's deadline
// window even at a 1ms interval — see #20266 for the flake pattern.
// Same-package access to sweepOnce lets us exercise the sweep logic
// deterministically without any wall-clock dependency.
func TestSessionTableCache_TTLSweepEvictsIdle(t *testing.T) {
clock := newFakeClock(time.Unix(1_700_000_000, 0))
var closeCount atomic.Int32
// Use a long sweepInterval so the background ticker never races the
// direct sweepOnce call below — the test asserts sweep behavior,
// not scheduler timing.
c := newSessionTableCache(1*time.Hour, 1*time.Hour, clock.now)
t.Cleanup(c.close)

// Open two handles, touch neither.
_ = c.getOrOpen("tbl:a", openClosing("tbl:a", &closeCount))
_ = c.getOrOpen("tbl:b", openClosing("tbl:b", &closeCount))

// Advance past TTL and drive a sweep synchronously.
clock.advance(2 * time.Hour)
c.sweepOnce()

c.mu.Lock()
n := len(c.entries)
c.mu.Unlock()
if n != 0 {
t.Errorf("entries remaining after TTL sweep = %d, want 0", n)
}
if got := closeCount.Load(); got != 2 {
t.Errorf("underlying Close called %d times on TTL evict, want 2", got)
}
}

// TestSessionTableCache_TouchDefersEviction pins that a ReadRow
// touch resets the idle timer — an entry touched every half-TTL
// stays alive indefinitely.
func TestSessionTableCache_TouchDefersEviction(t *testing.T) {
clock := newFakeClock(time.Unix(1_700_000_000, 0))
c := newTestSessionTableCache(t, 1*time.Hour, clock)

h := c.getOrOpen("tbl:t", openNoop("tbl:t")).(*sessionTableHandle)
// Every half-TTL, touch and step past a full TTL from the LAST
// touch. Each touch resets the clock reference so eviction never
// triggers.
for i := 0; i < 4; i++ {
clock.advance(30 * time.Minute)
_, _ = h.ReadRow(context.Background(), &btpb.SessionReadRowRequest{Key: []byte("r")})
}
// Drive a sweep directly; touch-driven lastAccess should keep the
// entry alive despite the clock advance.
c.sweepOnce()
c.mu.Lock()
_, present := c.entries["tbl:t"]
c.mu.Unlock()
if !present {
t.Error("touched entry got evicted; touch is not deferring eviction")
}
}

// TestSessionTableCache_CloseEvictsAll pins that closing the cache
// itself stops the sweeper and closes every remaining entry.
func TestSessionTableCache_CloseEvictsAll(t *testing.T) {
clock := newFakeClock(time.Unix(1_700_000_000, 0))
var closeCount atomic.Int32
c := newSessionTableCache(1*time.Hour, 1*time.Hour, clock.now)

_ = c.getOrOpen("tbl:a", openClosing("tbl:a", &closeCount))
_ = c.getOrOpen("tbl:b", openClosing("tbl:b", &closeCount))
_ = c.getOrOpen("mv:v", openClosing("mv:v", &closeCount))

c.close()

if got := closeCount.Load(); got != 3 {
t.Errorf("close(): underlying Close called %d times, want 3", got)
}
// close() is idempotent.
c.close()
}

// TestSessionTableCache_ClosedGate_SlowPathInsertNoLeak reproduces
// audit finding #5: without the closed-gate in getOrOpen's slow path,
// a caller whose openFn straddles cache.close() would leak the
// freshly-opened api (installed into a cache the sweeper has already
// stopped clearing). This test forces that interleaving via a
// synchronization channel on the openFn.
//
// Shape: caller A begins getOrOpen("k") on an empty cache. Fast-path
// misses. Slow-path calls openFn. openFn blocks until we signal it to
// return. Meanwhile close() runs on the cache (flips closed, walks the
// empty map). Then openFn returns. getOrOpen re-locks, sees closed,
// releases the fresh api itself, and returns nil. Underlying api's
// Close counter must reach 1.
func TestSessionTableCache_ClosedGate_SlowPathInsertNoLeak(t *testing.T) {
clock := newFakeClock(time.Unix(1_700_000_000, 0))
var closeCount atomic.Int32
c := newSessionTableCache(1*time.Hour, 1*time.Hour, clock.now)

release := make(chan struct{})
opened := make(chan struct{})
openFn := func() session.TableAPI {
close(opened)
<-release
return &closeCountingTable{noopSessionTable{key: "k"}, &closeCount}
}

var result session.TableAPI
done := make(chan struct{})
go func() {
result = c.getOrOpen("k", openFn)
close(done)
}()

// Wait until openFn is in-flight, THEN close the cache. This is the
// race window: openFn returns AFTER close() completed.
<-opened
c.close()
close(release)
<-done

if result != nil {
t.Errorf("getOrOpen returned non-nil after cache close: %v (want nil so TableShim falls back to classic)", result)
}
if got := closeCount.Load(); got != 1 {
t.Errorf("underlying api Close called %d times, want 1 (slow-path insert must release its api when cache is closed)", got)
}
// Cache map must be empty — the fresh api MUST NOT have been
// installed into an already-closed cache.
c.mu.Lock()
n := len(c.entries)
c.mu.Unlock()
if n != 0 {
t.Errorf("cache.entries has %d entries after close-race; want 0 (fresh handle must not be installed)", n)
}
}

// closeCountingTable is a noopSessionTable that atomically
// increments a counter on Close so tests can assert eviction actually
// called Close from any goroutine (including the cache sweeper).
type closeCountingTable struct {
noopSessionTable
counter *atomic.Int32
}

func (c *closeCountingTable) Close() error {
c.counter.Add(1)
return nil
}
Loading
Loading
0