Janady13/missionmatch / internal/platform/tenantpool/manager_test.go

1package tenantpool 2 3import ( 4 "context" 5 "errors" 6 "fmt" 7 "sync" 8 "sync/atomic" 9 "testing" 10 "time" 11 12 "github.com/google/uuid" 13 14 "github.com/missionworkforce/missionmatch/internal/platform/directory" 15 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 16) 17 18// clock is a manual time source so LRU and idle eviction are deterministic. 19type clock struct { 20 mu sync.Mutex 21 t time.Time 22} 23 24func (c *clock) now() time.Time { c.mu.Lock(); defer c.mu.Unlock(); return c.t } 25func (c *clock) advance(d time.Duration) { 26 c.mu.Lock() 27 c.t = c.t.Add(d) 28 c.mu.Unlock() 29} 30 31// fakeBuild stands in for the real open and verify path; it records build calls and per tenant closes. 32type fakeBuild struct { 33 mu sync.Mutex 34 calls int32 35 closes map[uuid.UUID]int 36 failFor map[uuid.UUID]error 37 delay time.Duration 38} 39 40func newFakeBuild() *fakeBuild { 41 return &fakeBuild{closes: map[uuid.UUID]int{}, failFor: map[uuid.UUID]error{}} 42} 43 44func (f *fakeBuild) build(_ context.Context, tc directory.TenantConn) (*opened, error) { 45 if f.delay > 0 { 46 time.Sleep(f.delay) 47 } 48 atomic.AddInt32(&f.calls, 1) 49 org := tc.OrganizationID 50 f.mu.Lock() 51 failErr := f.failFor[org] 52 f.mu.Unlock() 53 if failErr != nil { 54 return nil, failErr 55 } 56 return &opened{ 57 db: &tenant.DB{}, // nil pool; the registry never touches Pool, only routes db and calls close 58 close: func() { f.mu.Lock(); f.closes[org]++; f.mu.Unlock() }, 59 ref: tc.SecretRef, 60 }, nil 61} 62 63func (f *fakeBuild) closeCount(org uuid.UUID) int { 64 f.mu.Lock() 65 defer f.mu.Unlock() 66 return f.closes[org] 67} 68 69func mapLookup(tcs map[uuid.UUID]directory.TenantConn) func(context.Context, uuid.UUID) (directory.TenantConn, error) { 70 return func(_ context.Context, org uuid.UUID) (directory.TenantConn, error) { 71 tc, ok := tcs[org] 72 if !ok { 73 return directory.TenantConn{}, fmt.Errorf("%w: %s", directory.ErrUnknownTenant, org) 74 } 75 return tc, nil 76 } 77} 78 79func active(org uuid.UUID) directory.TenantConn { 80 return directory.TenantConn{OrganizationID: org, SecretRef: "ref-" + org.String()[:4], Status: directory.StatusActive} 81} 82 83func newTestManager(maxPools int, lookup func(context.Context, uuid.UUID) (directory.TenantConn, error), build func(context.Context, directory.TenantConn) (*opened, error)) *Manager { 84 return &Manager{ 85 lookup: lookup, 86 build: build, 87 maxPools: maxPools, 88 now: time.Now, 89 pools: map[uuid.UUID]*entry{}, 90 stop: make(chan struct{}), 91 } 92} 93 94func TestLazyOpenAndCacheHit(t *testing.T) { 95 org := uuid.New() 96 fb := newFakeBuild() 97 m := newTestManager(8, mapLookup(map[uuid.UUID]directory.TenantConn{org: active(org)}), fb.build) 98 ctx := context.Background() 99 100 _, rel1, err := m.Acquire(ctx, org) 101 if err != nil { 102 t.Fatalf("first acquire: %v", err) 103 } 104 _, rel2, err := m.Acquire(ctx, org) 105 if err != nil { 106 t.Fatalf("second acquire: %v", err) 107 } 108 if got := atomic.LoadInt32(&fb.calls); got != 1 { 109 t.Fatalf("a tenant pool must open exactly once and be cached, build calls = %d", got) 110 } 111 if m.Len() != 1 { 112 t.Fatalf("one live pool expected, got %d", m.Len()) 113 } 114 rel1() 115 rel2() 116} 117 118func TestUnknownTenantIsTyped(t *testing.T) { 119 m := newTestManager(8, mapLookup(map[uuid.UUID]directory.TenantConn{}), newFakeBuild().build) 120 _, _, err := m.Acquire(context.Background(), uuid.New()) 121 if !errors.Is(err, directory.ErrUnknownTenant) { 122 t.Fatalf("unknown tenant must be directory.ErrUnknownTenant, got %v", err) 123 } 124} 125 126func TestInactiveStatusRefuses(t *testing.T) { 127 org := uuid.New() 128 tc := active(org) 129 tc.Status = directory.StatusMigrating 130 m := newTestManager(8, mapLookup(map[uuid.UUID]directory.TenantConn{org: tc}), newFakeBuild().build) 131 _, _, err := m.Acquire(context.Background(), org) 132 if !errors.Is(err, ErrTenantUnavailable) { 133 t.Fatalf("a non active tenant must be ErrTenantUnavailable, got %v", err) 134 } 135} 136 137func TestUnreachableBuildIsTyped(t *testing.T) { 138 org := uuid.New() 139 fb := newFakeBuild() 140 fb.failFor[org] = fmt.Errorf("%w: dial refused", ErrTenantUnreachable) 141 m := newTestManager(8, mapLookup(map[uuid.UUID]directory.TenantConn{org: active(org)}), fb.build) 142 _, _, err := m.Acquire(context.Background(), org) 143 if !errors.Is(err, ErrTenantUnreachable) { 144 t.Fatalf("an unreachable tenant must be ErrTenantUnreachable, got %v", err) 145 } 146} 147 148func TestSingleflightOpensOnce(t *testing.T) { 149 org := uuid.New() 150 fb := newFakeBuild() 151 fb.delay = 20 * time.Millisecond 152 m := newTestManager(8, mapLookup(map[uuid.UUID]directory.TenantConn{org: active(org)}), fb.build) 153 ctx := context.Background() 154 155 var wg sync.WaitGroup 156 for i := 0; i < 16; i++ { 157 wg.Add(1) 158 go func() { 159 defer wg.Done() 160 _, rel, err := m.Acquire(ctx, org) 161 if err != nil { 162 t.Errorf("concurrent acquire: %v", err) 163 return 164 } 165 rel() 166 }() 167 } 168 wg.Wait() 169 if got := atomic.LoadInt32(&fb.calls); got != 1 { 170 t.Fatalf("singleflight must open exactly one pool under concurrency, build calls = %d", got) 171 } 172 if m.Len() != 1 { 173 t.Fatalf("one live pool expected, got %d", m.Len()) 174 } 175} 176 177func TestLRUEvictionClosesUnpinned(t *testing.T) { 178 a, b, c := uuid.New(), uuid.New(), uuid.New() 179 fb := newFakeBuild() 180 clk := &clock{t: time.Unix(1000, 0)} 181 m := newTestManager(2, mapLookup(map[uuid.UUID]directory.TenantConn{a: active(a), b: active(b), c: active(c)}), fb.build) 182 m.now = clk.now 183 ctx := context.Background() 184 185 clk.advance(time.Second) 186 _, relA, _ := m.Acquire(ctx, a) 187 relA() 188 clk.advance(time.Second) 189 _, relB, _ := m.Acquire(ctx, b) 190 relB() 191 clk.advance(time.Second) 192 _, relC, _ := m.Acquire(ctx, c) // over cap (3 > 2): evict the LRU unpinned, which is A 193 relC() 194 195 if m.Len() != 2 { 196 t.Fatalf("registry must stay at the cap, got %d", m.Len()) 197 } 198 if fb.closeCount(a) != 1 { 199 t.Fatalf("the least recently used pool (A) must be closed on eviction, closes = %d", fb.closeCount(a)) 200 } 201 if fb.closeCount(b) != 0 || fb.closeCount(c) != 0 { 202 t.Fatalf("B and C must remain open") 203 } 204} 205 206func TestPinnedPoolIsNeverEvicted(t *testing.T) { 207 a, b := uuid.New(), uuid.New() 208 fb := newFakeBuild() 209 m := newTestManager(1, mapLookup(map[uuid.UUID]directory.TenantConn{a: active(a), b: active(b)}), fb.build) 210 ctx := context.Background() 211 212 _, relA, _ := m.Acquire(ctx, a) // A pinned (not released) 213 _, relB, _ := m.Acquire(ctx, b) // over cap, but A is pinned so it cannot be evicted 214 if fb.closeCount(a) != 0 { 215 t.Fatalf("a pinned (in flight) pool must never be closed, A closes = %d", fb.closeCount(a)) 216 } 217 if m.Len() != 2 { 218 t.Fatalf("registry is temporarily over cap with two pinned pools, got %d", m.Len()) 219 } 220 relA() 221 relB() 222} 223 224func TestIdleSweepClosesQuietPools(t *testing.T) { 225 org := uuid.New() 226 fb := newFakeBuild() 227 clk := &clock{t: time.Unix(2000, 0)} 228 m := newTestManager(8, mapLookup(map[uuid.UUID]directory.TenantConn{org: active(org)}), fb.build) 229 m.idle = 10 * time.Second 230 m.now = clk.now 231 232 _, rel, _ := m.Acquire(context.Background(), org) 233 rel() 234 clk.advance(11 * time.Second) 235 m.sweepOnce() 236 if m.Len() != 0 { 237 t.Fatalf("an idle pool past the TTL must be swept, live = %d", m.Len()) 238 } 239 if fb.closeCount(org) != 1 { 240 t.Fatalf("a swept pool must be closed, closes = %d", fb.closeCount(org)) 241 } 242} 243 244func TestCloseClosesAllAndRefuses(t *testing.T) { 245 a, b := uuid.New(), uuid.New() 246 fb := newFakeBuild() 247 m := newTestManager(8, mapLookup(map[uuid.UUID]directory.TenantConn{a: active(a), b: active(b)}), fb.build) 248 ctx := context.Background() 249 _, relA, _ := m.Acquire(ctx, a) 250 relA() 251 _, relB, _ := m.Acquire(ctx, b) 252 relB() 253 254 m.Close() 255 if fb.closeCount(a) != 1 || fb.closeCount(b) != 1 { 256 t.Fatalf("Close must close every live pool, A=%d B=%d", fb.closeCount(a), fb.closeCount(b)) 257 } 258 if _, _, err := m.Acquire(ctx, a); !errors.Is(err, ErrClosed) { 259 t.Fatalf("a closed manager must refuse Acquire, got %v", err) 260 } 261} 262