Address review: close fallback KernelBackend, test TTL expiry

- ThriftBackend: track the KernelBackend(s) created for Reyden (KP001) fallback
  and close them in close(), so the process-global log-bridge onLevelChange
  listener installed by connect() is released instead of leaking on every
  recovery. Add a createKernelBackend() seam so tests can inject a fake without
  the native binding, plus a test that close() releases the fallback backend.
- ReydenWarehouseCache test: add a TTL-expiry test (sinon fake timers) covering
  the 6h boundary and opportunistic eviction on access.

Co-authored-by: Isaac <no-reply@databricks.com>
Signed-off-by: Rahul Singhal <rahul.singhal@databricks.com>

Author: rahuls-db

lib/thrift-backend/ThriftBackend.ts

@@ -37,6 +37,11 @@ export default class ThriftBackend implements IBackend {
 
   private connectionOptions?: ConnectionOptions;
 
+  // KernelBackend(s) created for Reyden (KP001) fallback. Tracked so their
+  // process-global log-bridge listeners are released on close() — otherwise each
+  // fallback session would leak an onLevelChange listener for the process lifetime.
+  private fallbackKernelBackends: KernelBackend[] = [];
+
   constructor({ context, onConnectionEvent }: ThriftBackendOptions) {
     this.context = context;
     this.onConnectionEvent = onConnectionEvent;
@@ -184,14 +189,24 @@ export default class ThriftBackend implements IBackend {
     const logger = this.context.getLogger();
     logger.log(LogLevel.debug, 'Reyden: opening session via KernelBackend (SEA)');
 
-    // Create a new KernelBackend instance and connect/open
-    const kernelBackend = new KernelBackend({ context: this.context });
+    // Create a KernelBackend and connect/open. Track it so close() releases the
+    // log-bridge listener that connect() installs.
+    const kernelBackend = this.createKernelBackend();
+    this.fallbackKernelBackends.push(kernelBackend);
     await kernelBackend.connect(this.connectionOptions);
     return kernelBackend.openSession(request);
   }
 
+  // Seam so tests can inject a fake KernelBackend without the native binding.
+  protected createKernelBackend(): KernelBackend {
+    return new KernelBackend({ context: this.context });
+  }
+
   public async close(): Promise<void> {
-    // DBSQLClient owns the connection lifecycle and clears its own state
-    // (connectionProvider, authProvider, thrift client) after this returns.
+    // Release the process-global log-bridge listener(s) held by any Reyden-fallback
+    // KernelBackend. DBSQLClient owns the rest of the connection lifecycle and clears
+    // its own state (connectionProvider, authProvider, thrift client) after this returns.
+    await Promise.all(this.fallbackKernelBackends.map((backend) => backend.close()));
+    this.fallbackKernelBackends = [];
   }
 }

tests/unit/thrift-backend/ReydenThriftRecovery.test.ts

@@ -1,4 +1,5 @@
 import { expect } from 'chai';
+import sinon from 'sinon';
 import reydenCache from '../../../lib/ReydenWarehouseCache';
 import ThriftBackend from '../../../lib/thrift-backend/ThriftBackend';
 import StatusError from '../../../lib/errors/StatusError';
@@ -97,6 +98,36 @@ describe('Reyden Warehouse Cache', () => {
       expect(reydenCache.isKnownReyden('host1.com', 'warehouse-1')).to.be.undefined;
     });
   });
+
+  describe('Cache TTL Expiry', () => {
+    let clock: sinon.SinonFakeTimers;
+
+    beforeEach(() => {
+      clock = sinon.useFakeTimers();
+    });
+
+    afterEach(() => {
+      clock.restore();
+    });
+
+    it('keeps an entry until the 6h TTL, then evicts it on access', () => {
+      const host = 'example.com';
+      const warehouseId = 'warehouse-ttl';
+      const sixHoursMs = 6 * 60 * 60 * 1000;
+
+      reydenCache.markReyden(host, warehouseId);
+      expect(reydenCache.isKnownReyden(host, warehouseId)).to.be.true;
+
+      // At exactly the TTL boundary the entry is still valid (strict >).
+      clock.tick(sixHoursMs);
+      expect(reydenCache.isKnownReyden(host, warehouseId)).to.be.true;
+
+      // One tick past the TTL: expired, evicted on access.
+      clock.tick(1);
+      expect(reydenCache.isKnownReyden(host, warehouseId)).to.be.undefined;
+      expect(reydenCache.size()).to.equal(0);
+    });
+  });
 });
 
 describe('StatusError SQLSTATE Support', () => {

tests/unit/thrift-backend/ReydenThriftRecoveryOrchestration.test.ts

@@ -125,4 +125,24 @@ describe('Reyden Thrift Auto-Recovery — Orchestration', () => {
     expect(thrown).to.equal(kernelError);
     expect(thrown.cause).to.equal(thriftError);
   });
+
+  it('closes the fallback KernelBackend on close(), releasing its log-bridge listener', async () => {
+    const backend = makeBackend();
+    const fakeKernel = {
+      connect: sandbox.stub().resolves(),
+      openSession: sandbox.stub().resolves({ marker: 'kernel-session' } as any),
+      close: sandbox.stub().resolves(),
+    };
+    sandbox.stub(backend as any, 'openSessionWithThrift').rejects(kp001Error());
+    // Inject the fake via the createKernelBackend seam and let the REAL
+    // openSessionWithKernelBackend run (connect + track), so close() must release it.
+    sandbox.stub(backend as any, 'createKernelBackend').returns(fakeKernel as any);
+
+    await backend.openSession({} as any);
+    expect(fakeKernel.connect.calledOnce).to.be.true;
+    expect(fakeKernel.close.called).to.be.false; // still open
+
+    await backend.close();
+    expect(fakeKernel.close.calledOnce).to.be.true; // released on ThriftBackend.close()
+  });
 });