Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2026514404 |
@@ -1,87 +0,0 @@
|
|||||||
// Package flag implementiert Core LIC-02: einen Feature-Flag-Dienst mit
|
|
||||||
// Strategien (global an/aus, Prozentsatz, Tenant-Zielgruppe) als Kernfunktion
|
|
||||||
// des Core-Dienstes selbst — keine zusaetzliche Infrastruktur (Unleash-Server
|
|
||||||
// + eigene DB), siehe "bewusst vermeiden" im LIC-02-Ticket.
|
|
||||||
package flag
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"errors"
|
|
||||||
"fmt"
|
|
||||||
"hash/fnv"
|
|
||||||
|
|
||||||
"github.com/jackc/pgx/v5"
|
|
||||||
"github.com/jackc/pgx/v5/pgxpool"
|
|
||||||
)
|
|
||||||
|
|
||||||
var ErrNotFound = errors.New("flag: nicht gefunden")
|
|
||||||
|
|
||||||
// Flag ist die zentrale Definition — Auswertung (Evaluate) ist bewusst davon
|
|
||||||
// getrennt (Unleash-Prinzip: Flag-Verwaltung vs. Flag-Auswertung).
|
|
||||||
type Flag struct {
|
|
||||||
Key string
|
|
||||||
Enabled bool
|
|
||||||
RolloutPercentage int
|
|
||||||
TargetTenantSlugs []string
|
|
||||||
}
|
|
||||||
|
|
||||||
// Store ist die Verwaltungsseite (Admin): Flags definieren/lesen.
|
|
||||||
type Store struct {
|
|
||||||
pool *pgxpool.Pool
|
|
||||||
}
|
|
||||||
|
|
||||||
func NewStore(pool *pgxpool.Pool) *Store {
|
|
||||||
return &Store{pool: pool}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *Store) Set(ctx context.Context, f Flag) error {
|
|
||||||
if f.TargetTenantSlugs == nil {
|
|
||||||
f.TargetTenantSlugs = []string{} // pgx uebertraegt ein nil-Slice sonst als SQL NULL statt leerem Array.
|
|
||||||
}
|
|
||||||
_, err := s.pool.Exec(ctx, `
|
|
||||||
INSERT INTO feature_flags (key, enabled, rollout_percentage, target_tenant_slugs, updated_at)
|
|
||||||
VALUES ($1, $2, $3, $4, now())
|
|
||||||
ON CONFLICT (key) DO UPDATE SET
|
|
||||||
enabled = $2, rollout_percentage = $3, target_tenant_slugs = $4, updated_at = now()
|
|
||||||
`, f.Key, f.Enabled, f.RolloutPercentage, f.TargetTenantSlugs)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("flag speichern: %w", err)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *Store) Get(ctx context.Context, key string) (Flag, error) {
|
|
||||||
var f Flag
|
|
||||||
row := s.pool.QueryRow(ctx, `
|
|
||||||
SELECT key, enabled, rollout_percentage, target_tenant_slugs
|
|
||||||
FROM feature_flags WHERE key = $1
|
|
||||||
`, key)
|
|
||||||
if err := row.Scan(&f.Key, &f.Enabled, &f.RolloutPercentage, &f.TargetTenantSlugs); err != nil {
|
|
||||||
if errors.Is(err, pgx.ErrNoRows) {
|
|
||||||
return Flag{}, ErrNotFound
|
|
||||||
}
|
|
||||||
return Flag{}, fmt.Errorf("flag lesen: %w", err)
|
|
||||||
}
|
|
||||||
return f, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// evaluate wendet die Strategien in fester Reihenfolge an: globaler
|
|
||||||
// An/Aus-Schalter zuerst, dann Tenant-Zielgruppe, dann Prozentsatz-Rollout.
|
|
||||||
// Ein unbekannter/nicht getroffener Fall ergibt false — Fail-Safe-Default,
|
|
||||||
// kein Feature wird versehentlich aktiv.
|
|
||||||
func evaluate(f Flag, tenantSlug string) bool {
|
|
||||||
if f.Enabled {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
for _, target := range f.TargetTenantSlugs {
|
|
||||||
if target == tenantSlug {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if f.RolloutPercentage > 0 {
|
|
||||||
h := fnv.New32a()
|
|
||||||
_, _ = h.Write([]byte(f.Key + "|" + tenantSlug))
|
|
||||||
return int(h.Sum32()%100) < f.RolloutPercentage
|
|
||||||
}
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
@@ -1,43 +0,0 @@
|
|||||||
package flag
|
|
||||||
|
|
||||||
import "testing"
|
|
||||||
|
|
||||||
func TestEvaluate_GlobalEnabled(t *testing.T) {
|
|
||||||
f := Flag{Key: "k", Enabled: true}
|
|
||||||
if !evaluate(f, "irgendein-tenant") {
|
|
||||||
t.Fatal("global aktiviertes flag sollte fuer jeden tenant true liefern")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Akzeptanzkriterium 1 + Pruefung 2: Zielgruppen-Strategie.
|
|
||||||
func TestEvaluate_TargetTenantStrategy(t *testing.T) {
|
|
||||||
f := Flag{Key: "k", Enabled: false, TargetTenantSlugs: []string{"acme"}}
|
|
||||||
if !evaluate(f, "acme") {
|
|
||||||
t.Fatal("erwartet true fuer tenant in zielgruppe")
|
|
||||||
}
|
|
||||||
if evaluate(f, "globex") {
|
|
||||||
t.Fatal("erwartet false fuer tenant ausserhalb der zielgruppe")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestEvaluate_RolloutPercentageBoundaries(t *testing.T) {
|
|
||||||
full := Flag{Key: "k", RolloutPercentage: 100}
|
|
||||||
if !evaluate(full, "beliebiger-tenant-1") || !evaluate(full, "beliebiger-tenant-2") {
|
|
||||||
t.Fatal("100% rollout sollte immer true liefern")
|
|
||||||
}
|
|
||||||
|
|
||||||
none := Flag{Key: "k", RolloutPercentage: 0}
|
|
||||||
if evaluate(none, "beliebiger-tenant") {
|
|
||||||
t.Fatal("0% rollout ohne enabled/zielgruppe sollte false liefern")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestEvaluate_RolloutIsDeterministicPerTenant(t *testing.T) {
|
|
||||||
f := Flag{Key: "k", RolloutPercentage: 50}
|
|
||||||
first := evaluate(f, "stabiler-tenant")
|
|
||||||
for i := 0; i < 5; i++ {
|
|
||||||
if evaluate(f, "stabiler-tenant") != first {
|
|
||||||
t.Fatal("rollout-auswertung sollte fuer denselben tenant/key stabil sein")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,87 +0,0 @@
|
|||||||
package flag
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"log/slog"
|
|
||||||
"sync"
|
|
||||||
"time"
|
|
||||||
)
|
|
||||||
|
|
||||||
// DefaultCacheTTL ist die dokumentierte Cache-Invalidierungszeit
|
|
||||||
// (Akzeptanzkriterium 2/3): eine Aenderung wirkt spaetestens nach dieser
|
|
||||||
// Zeit auf allen Core-Instanzen, ohne dass ein Dienst neu gestartet werden
|
|
||||||
// muss (Akzeptanzkriterium 3).
|
|
||||||
const DefaultCacheTTL = 5 * time.Second
|
|
||||||
|
|
||||||
type cacheEntry struct {
|
|
||||||
flag Flag
|
|
||||||
expiresAt time.Time
|
|
||||||
}
|
|
||||||
|
|
||||||
// Service ist die Auswertungsseite (SDK/Client-Analogon zu Unleash) mit
|
|
||||||
// lokalem TTL-Cache. Bewusst getrennt von Store (Verwaltung).
|
|
||||||
type Service struct {
|
|
||||||
store *Store
|
|
||||||
ttl time.Duration
|
|
||||||
|
|
||||||
mu sync.RWMutex
|
|
||||||
cache map[string]cacheEntry
|
|
||||||
}
|
|
||||||
|
|
||||||
func NewService(store *Store, ttl time.Duration) *Service {
|
|
||||||
if ttl <= 0 {
|
|
||||||
ttl = DefaultCacheTTL
|
|
||||||
}
|
|
||||||
return &Service{store: store, ttl: ttl, cache: make(map[string]cacheEntry)}
|
|
||||||
}
|
|
||||||
|
|
||||||
// IsEnabled wertet ein Flag fuer einen Tenant aus. Liefert IMMER einen
|
|
||||||
// bool ohne Fehlerwert — ein nicht erreichbarer Flag-Dienst darf abhaengige
|
|
||||||
// Aufrufer nicht zum Absturz bringen oder zu Fehlerbehandlungscode zwingen,
|
|
||||||
// der leicht vergessen wird (Akzeptanzkriterium 3 / Pruefung 3: dokumentiertes
|
|
||||||
// Fallback-Verhalten = false, ggf. aus dem zuletzt bekannten Zwischenspeicher).
|
|
||||||
func (s *Service) IsEnabled(ctx context.Context, tenantSlug, key string) bool {
|
|
||||||
f, ok := s.resolve(ctx, key)
|
|
||||||
if !ok {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
return evaluate(f, tenantSlug)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *Service) resolve(ctx context.Context, key string) (Flag, bool) {
|
|
||||||
s.mu.RLock()
|
|
||||||
entry, exists := s.cache[key]
|
|
||||||
fresh := exists && time.Now().Before(entry.expiresAt)
|
|
||||||
s.mu.RUnlock()
|
|
||||||
if fresh {
|
|
||||||
return entry.flag, true
|
|
||||||
}
|
|
||||||
|
|
||||||
f, err := s.store.Get(ctx, key)
|
|
||||||
if err != nil {
|
|
||||||
if exists {
|
|
||||||
slog.Warn("feature-flag-dienst nicht erreichbar, nutze zwischengespeicherten stand",
|
|
||||||
"flag_key", key, "error", err)
|
|
||||||
return entry.flag, true
|
|
||||||
}
|
|
||||||
slog.Warn("feature-flag-dienst nicht erreichbar, kein zwischengespeicherter stand vorhanden, fallback: deaktiviert",
|
|
||||||
"flag_key", key, "error", err)
|
|
||||||
return Flag{}, false
|
|
||||||
}
|
|
||||||
|
|
||||||
s.mu.Lock()
|
|
||||||
s.cache[key] = cacheEntry{flag: f, expiresAt: time.Now().Add(s.ttl)}
|
|
||||||
s.mu.Unlock()
|
|
||||||
return f, true
|
|
||||||
}
|
|
||||||
|
|
||||||
// Invalidate erzwingt beim naechsten IsEnabled-Aufruf ein sofortiges Neuladen
|
|
||||||
// aus der Datenbank statt auf den TTL-Ablauf zu warten — wird nach Store.Set
|
|
||||||
// auf derselben Instanz aufgerufen, damit der Schreiber die eigene Aenderung
|
|
||||||
// ohne Wartezeit sieht. Andere Core-Instanzen sehen sie spaetestens nach
|
|
||||||
// DefaultCacheTTL (siehe Akzeptanzkriterium 3).
|
|
||||||
func (s *Service) Invalidate(key string) {
|
|
||||||
s.mu.Lock()
|
|
||||||
delete(s.cache, key)
|
|
||||||
s.mu.Unlock()
|
|
||||||
}
|
|
||||||
@@ -1,179 +0,0 @@
|
|||||||
package flag
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"os"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/jackc/pgx/v5/pgxpool"
|
|
||||||
)
|
|
||||||
|
|
||||||
func setupFlagStoreTest(t *testing.T) (*Store, func()) {
|
|
||||||
t.Helper()
|
|
||||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
|
||||||
if adminDSN == "" {
|
|
||||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
|
||||||
}
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
pool, err := pgxpool.New(ctx, adminDSN)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("pool: %v", err)
|
|
||||||
}
|
|
||||||
if _, err := pool.Exec(ctx, `
|
|
||||||
CREATE TABLE IF NOT EXISTS feature_flags (
|
|
||||||
key TEXT PRIMARY KEY,
|
|
||||||
enabled BOOLEAN NOT NULL DEFAULT false,
|
|
||||||
rollout_percentage INT NOT NULL DEFAULT 0,
|
|
||||||
target_tenant_slugs TEXT[] NOT NULL DEFAULT '{}',
|
|
||||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
|
||||||
)`); err != nil {
|
|
||||||
t.Fatalf("schema: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
cleanup := func() {
|
|
||||||
_, _ = pool.Exec(ctx, `DELETE FROM feature_flags WHERE key LIKE 'test\_%' ESCAPE '\'`)
|
|
||||||
pool.Close()
|
|
||||||
}
|
|
||||||
return NewStore(pool), cleanup
|
|
||||||
}
|
|
||||||
|
|
||||||
// Akzeptanzkriterium 1 + Pruefung 2: Zielgruppen-Strategie liefert im Test
|
|
||||||
// die erwartete Auswertung.
|
|
||||||
func TestService_TargetTenantStrategy(t *testing.T) {
|
|
||||||
store, cleanup := setupFlagStoreTest(t)
|
|
||||||
defer cleanup()
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
if err := store.Set(ctx, Flag{Key: "test_target_flag", TargetTenantSlugs: []string{"acme"}}); err != nil {
|
|
||||||
t.Fatalf("set: %v", err)
|
|
||||||
}
|
|
||||||
svc := NewService(store, time.Hour)
|
|
||||||
|
|
||||||
if !svc.IsEnabled(ctx, "acme", "test_target_flag") {
|
|
||||||
t.Fatal("erwartet true fuer tenant in zielgruppe")
|
|
||||||
}
|
|
||||||
if svc.IsEnabled(ctx, "globex", "test_target_flag") {
|
|
||||||
t.Fatal("erwartet false fuer tenant ausserhalb der zielgruppe")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Akzeptanzkriterium 2 + 3 + Pruefung 1: Flag-Aenderung wirkt innerhalb der
|
|
||||||
// dokumentierten Cache-Invalidierungszeit, automatisiert gemessen.
|
|
||||||
func TestService_CacheInvalidationTiming(t *testing.T) {
|
|
||||||
store, cleanup := setupFlagStoreTest(t)
|
|
||||||
defer cleanup()
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
const ttl = 150 * time.Millisecond
|
|
||||||
if err := store.Set(ctx, Flag{Key: "test_ttl_flag", Enabled: false}); err != nil {
|
|
||||||
t.Fatalf("set: %v", err)
|
|
||||||
}
|
|
||||||
svc := NewService(store, ttl)
|
|
||||||
|
|
||||||
if svc.IsEnabled(ctx, "acme", "test_ttl_flag") {
|
|
||||||
t.Fatal("erwartet false vor der aenderung")
|
|
||||||
}
|
|
||||||
|
|
||||||
// Aenderung "auf einer anderen instanz" simulieren: direkt ueber den
|
|
||||||
// Store, ohne svc.Invalidate aufzurufen.
|
|
||||||
changedAt := time.Now()
|
|
||||||
if err := store.Set(ctx, Flag{Key: "test_ttl_flag", Enabled: true}); err != nil {
|
|
||||||
t.Fatalf("set: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Sofort danach sollte der Cache noch den alten Stand liefern.
|
|
||||||
if svc.IsEnabled(ctx, "acme", "test_ttl_flag") {
|
|
||||||
t.Fatal("cache haette den alten (false) stand liefern sollen, direkt nach der aenderung")
|
|
||||||
}
|
|
||||||
|
|
||||||
deadline := changedAt.Add(ttl + 100*time.Millisecond)
|
|
||||||
for time.Now().Before(deadline) {
|
|
||||||
if svc.IsEnabled(ctx, "acme", "test_ttl_flag") {
|
|
||||||
elapsed := time.Since(changedAt)
|
|
||||||
t.Logf("aenderung wurde nach %s wirksam (ziel: innerhalb %s + toleranz)", elapsed, ttl)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
time.Sleep(10 * time.Millisecond)
|
|
||||||
}
|
|
||||||
t.Fatalf("aenderung wurde nicht innerhalb von %s wirksam", deadline.Sub(changedAt))
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestService_InvalidateForcesImmediateRefresh(t *testing.T) {
|
|
||||||
store, cleanup := setupFlagStoreTest(t)
|
|
||||||
defer cleanup()
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
if err := store.Set(ctx, Flag{Key: "test_invalidate_flag", Enabled: false}); err != nil {
|
|
||||||
t.Fatalf("set: %v", err)
|
|
||||||
}
|
|
||||||
svc := NewService(store, time.Hour) // lange TTL, damit Invalidate den unterschied macht
|
|
||||||
_ = svc.IsEnabled(ctx, "acme", "test_invalidate_flag")
|
|
||||||
|
|
||||||
if err := store.Set(ctx, Flag{Key: "test_invalidate_flag", Enabled: true}); err != nil {
|
|
||||||
t.Fatalf("set: %v", err)
|
|
||||||
}
|
|
||||||
svc.Invalidate("test_invalidate_flag")
|
|
||||||
|
|
||||||
if !svc.IsEnabled(ctx, "acme", "test_invalidate_flag") {
|
|
||||||
t.Fatal("erwartet sofort sichtbaren neuen stand nach Invalidate")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Akzeptanzkriterium 3 + Pruefung 3: Ausfall des Flag-Dienstes fuehrt zu
|
|
||||||
// dokumentiertem Fallback-Verhalten, nicht zum Absturz.
|
|
||||||
func TestService_FallsBackOnStoreFailure(t *testing.T) {
|
|
||||||
store, cleanup := setupFlagStoreTest(t)
|
|
||||||
defer cleanup()
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
if err := store.Set(ctx, Flag{Key: "test_fallback_flag", Enabled: true}); err != nil {
|
|
||||||
t.Fatalf("set: %v", err)
|
|
||||||
}
|
|
||||||
svc := NewService(store, time.Hour)
|
|
||||||
|
|
||||||
// Cache vorwaermen, waehrend die DB noch erreichbar ist.
|
|
||||||
if !svc.IsEnabled(ctx, "acme", "test_fallback_flag") {
|
|
||||||
t.Fatal("erwartet true bei funktionierender db")
|
|
||||||
}
|
|
||||||
|
|
||||||
brokenPool, err := pgxpool.New(ctx, "postgresql://nonexistent-host-fuer-test:5432/x?connect_timeout=1")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("broken pool erstellen (sollte nicht sofort verbinden): %v", err)
|
|
||||||
}
|
|
||||||
brokenStore := NewStore(brokenPool)
|
|
||||||
|
|
||||||
svcWithCache := NewService(brokenStore, time.Nanosecond) // TTL sofort abgelaufen, erzwingt reload-versuch
|
|
||||||
svcWithCache.mu.Lock()
|
|
||||||
svcWithCache.cache["test_fallback_flag"] = cacheEntry{
|
|
||||||
flag: Flag{Key: "test_fallback_flag", Enabled: true},
|
|
||||||
expiresAt: time.Now().Add(-time.Hour), // bereits abgelaufen
|
|
||||||
}
|
|
||||||
svcWithCache.mu.Unlock()
|
|
||||||
|
|
||||||
func() {
|
|
||||||
defer func() {
|
|
||||||
if r := recover(); r != nil {
|
|
||||||
t.Fatalf("IsEnabled hat gepanict statt einen fallback zu liefern: %v", r)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
if !svcWithCache.IsEnabled(ctx, "acme", "test_fallback_flag") {
|
|
||||||
t.Fatal("erwartet fallback auf zwischengespeicherten (true) stand bei db-ausfall")
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
// Voellig frischer Dienst ohne jeglichen cache + kaputte db -> sicherer
|
|
||||||
// default false, kein absturz.
|
|
||||||
freshSvc := NewService(brokenStore, time.Hour)
|
|
||||||
func() {
|
|
||||||
defer func() {
|
|
||||||
if r := recover(); r != nil {
|
|
||||||
t.Fatalf("IsEnabled hat gepanict: %v", r)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
if freshSvc.IsEnabled(ctx, "acme", "test_fallback_flag") {
|
|
||||||
t.Fatal("erwartet fail-safe false ohne cache und mit kaputter db")
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
@@ -0,0 +1,136 @@
|
|||||||
|
package tenant
|
||||||
|
|
||||||
|
import (
|
||||||
|
"container/list"
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"sync"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ErrMissingTenantContext wird geliefert, wenn keine Tenant-Kennung
|
||||||
|
// uebergeben wurde — es gibt bewusst keinen impliziten Default-Tenant
|
||||||
|
// (TEN-06 Akzeptanzkriterium 3).
|
||||||
|
var ErrMissingTenantContext = errors.New("tenant: kein tenant-kontext angegeben")
|
||||||
|
|
||||||
|
// ErrUnknownTenant wird geliefert, wenn die Tenant-Kennung in der Registry
|
||||||
|
// nicht existiert.
|
||||||
|
var ErrUnknownTenant = errors.New("tenant: unbekannter tenant")
|
||||||
|
|
||||||
|
// Router loest den Tenant-Kontext (Slug, aus dem JWT-Claim von API-05) in
|
||||||
|
// eine wiederverwendbare Verbindung zur richtigen Tenant-Datenbank auf.
|
||||||
|
// Ein LRU-verwalteter Cache begrenzt die Zahl gleichzeitig offener
|
||||||
|
// pgxpool.Pool-Instanzen, damit die Zahl offener Postgres-Verbindungen NICHT
|
||||||
|
// linear mit der Mandantenzahl waechst (Akzeptanzkriterium 2).
|
||||||
|
type Router struct {
|
||||||
|
registry *Registry
|
||||||
|
maxOpen int
|
||||||
|
|
||||||
|
mu sync.Mutex
|
||||||
|
order *list.List // vorne = zuletzt benutzt
|
||||||
|
items map[string]*list.Element // slug -> element mit *routerEntry
|
||||||
|
}
|
||||||
|
|
||||||
|
type routerEntry struct {
|
||||||
|
slug string
|
||||||
|
pool *pgxpool.Pool
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewRouter(registry *Registry, maxOpen int) *Router {
|
||||||
|
if maxOpen < 1 {
|
||||||
|
maxOpen = 1
|
||||||
|
}
|
||||||
|
return &Router{
|
||||||
|
registry: registry,
|
||||||
|
maxOpen: maxOpen,
|
||||||
|
order: list.New(),
|
||||||
|
items: make(map[string]*list.Element),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Resolve liefert einen wiederverwendeten Pool fuer den angegebenen Tenant.
|
||||||
|
// Ist der Tenant bereits im Cache, wird KEINE neue Verbindung aufgebaut
|
||||||
|
// (Akzeptanzkriterium 2 / Pruefung 3).
|
||||||
|
func (r *Router) Resolve(ctx context.Context, tenantSlug string) (*pgxpool.Pool, error) {
|
||||||
|
if tenantSlug == "" {
|
||||||
|
return nil, ErrMissingTenantContext
|
||||||
|
}
|
||||||
|
|
||||||
|
r.mu.Lock()
|
||||||
|
if el, ok := r.items[tenantSlug]; ok {
|
||||||
|
r.order.MoveToFront(el)
|
||||||
|
pool := el.Value.(*routerEntry).pool
|
||||||
|
r.mu.Unlock()
|
||||||
|
return pool, nil
|
||||||
|
}
|
||||||
|
r.mu.Unlock()
|
||||||
|
|
||||||
|
// Registry-Lookup und Verbindungsaufbau bewusst ausserhalb des Locks,
|
||||||
|
// damit ein langsamer Verbindungsaufbau nicht alle anderen Tenants blockiert.
|
||||||
|
t, err := r.registry.GetBySlug(ctx, tenantSlug)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("%w: %s", ErrUnknownTenant, tenantSlug)
|
||||||
|
}
|
||||||
|
|
||||||
|
pool, err := pgxpool.New(ctx, t.DBDSN)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("verbindung zu tenant %q aufbauen: %w", tenantSlug, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
r.mu.Lock()
|
||||||
|
defer r.mu.Unlock()
|
||||||
|
|
||||||
|
// Zwischen Unlock oben und hier koennte ein paralleler Aufruf denselben
|
||||||
|
// Tenant bereits eingefuegt haben — dann die eigene, ueberzaehlige
|
||||||
|
// Verbindung wieder schliessen und die vorhandene verwenden.
|
||||||
|
if el, ok := r.items[tenantSlug]; ok {
|
||||||
|
r.order.MoveToFront(el)
|
||||||
|
existing := el.Value.(*routerEntry).pool
|
||||||
|
pool.Close()
|
||||||
|
return existing, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
el := r.order.PushFront(&routerEntry{slug: tenantSlug, pool: pool})
|
||||||
|
r.items[tenantSlug] = el
|
||||||
|
|
||||||
|
if r.order.Len() > r.maxOpen {
|
||||||
|
r.evictOldest()
|
||||||
|
}
|
||||||
|
|
||||||
|
return pool, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// evictOldest schliesst den am laengsten nicht genutzten Pool. Muss mit
|
||||||
|
// gehaltenem r.mu aufgerufen werden.
|
||||||
|
func (r *Router) evictOldest() {
|
||||||
|
oldest := r.order.Back()
|
||||||
|
if oldest == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
entry := oldest.Value.(*routerEntry)
|
||||||
|
r.order.Remove(oldest)
|
||||||
|
delete(r.items, entry.slug)
|
||||||
|
entry.pool.Close()
|
||||||
|
}
|
||||||
|
|
||||||
|
// OpenCount liefert die aktuelle Zahl offen gehaltener Tenant-Pools —
|
||||||
|
// dient Tests/Monitoring, um AC2 nachzuweisen.
|
||||||
|
func (r *Router) OpenCount() int {
|
||||||
|
r.mu.Lock()
|
||||||
|
defer r.mu.Unlock()
|
||||||
|
return r.order.Len()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Close schliesst alle offen gehaltenen Tenant-Pools, z.B. beim
|
||||||
|
// Herunterfahren des Core-Prozesses.
|
||||||
|
func (r *Router) Close() {
|
||||||
|
r.mu.Lock()
|
||||||
|
defer r.mu.Unlock()
|
||||||
|
for el := r.order.Front(); el != nil; el = el.Next() {
|
||||||
|
el.Value.(*routerEntry).pool.Close()
|
||||||
|
}
|
||||||
|
r.order.Init()
|
||||||
|
r.items = make(map[string]*list.Element)
|
||||||
|
}
|
||||||
@@ -0,0 +1,160 @@
|
|||||||
|
package tenant
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
|
)
|
||||||
|
|
||||||
|
func newTestRouterSetup(t *testing.T, tenantCount int) (*Router, []Tenant, func()) {
|
||||||
|
t.Helper()
|
||||||
|
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||||
|
if adminDSN == "" {
|
||||||
|
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||||
|
}
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
adminPool, err := pgxpool.New(ctx, adminDSN)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("admin pool: %v", err)
|
||||||
|
}
|
||||||
|
registryPool, err := pgxpool.New(ctx, adminDSN)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("registry pool: %v", err)
|
||||||
|
}
|
||||||
|
if _, err := registryPool.Exec(ctx, `
|
||||||
|
CREATE TABLE IF NOT EXISTS tenants (
|
||||||
|
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||||
|
slug TEXT NOT NULL UNIQUE,
|
||||||
|
name TEXT NOT NULL,
|
||||||
|
db_name TEXT NOT NULL UNIQUE,
|
||||||
|
db_dsn TEXT NOT NULL,
|
||||||
|
status TEXT NOT NULL DEFAULT 'active',
|
||||||
|
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||||
|
)`); err != nil {
|
||||||
|
t.Fatalf("registry-schema: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
registry := NewRegistry(registryPool)
|
||||||
|
dsnTemplate := strings.Replace(adminDSN, "/postgres?", "/%s?", 1)
|
||||||
|
provisioner := NewProvisioner(adminPool, registry, dsnTemplate)
|
||||||
|
|
||||||
|
var tenants []Tenant
|
||||||
|
var slugs []string
|
||||||
|
for i := 0; i < tenantCount; i++ {
|
||||||
|
slug := fmt.Sprintf("router_t%d", i)
|
||||||
|
slugs = append(slugs, slug)
|
||||||
|
tn, err := provisioner.Provision(ctx, slug, slug)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("provision %s: %v", slug, err)
|
||||||
|
}
|
||||||
|
tenants = append(tenants, tn)
|
||||||
|
}
|
||||||
|
|
||||||
|
router := NewRouter(registry, 2) // klein gewaehlt, um Eviction im Test zu erzwingen
|
||||||
|
|
||||||
|
cleanup := func() {
|
||||||
|
router.Close()
|
||||||
|
for _, slug := range slugs {
|
||||||
|
_, _ = adminPool.Exec(ctx, fmt.Sprintf(`DROP DATABASE IF EXISTS %q`, dbNameForSlug(slug)))
|
||||||
|
}
|
||||||
|
_, _ = registryPool.Exec(ctx, `DELETE FROM tenants WHERE slug = ANY($1)`, slugs)
|
||||||
|
registryPool.Close()
|
||||||
|
adminPool.Close()
|
||||||
|
}
|
||||||
|
return router, tenants, cleanup
|
||||||
|
}
|
||||||
|
|
||||||
|
// Akzeptanzkriterium 1: Verbindung wird zuverlaessig anhand des Tenant-Kontexts aufgeloest.
|
||||||
|
func TestRouter_ResolvesCorrectTenantDatabase(t *testing.T) {
|
||||||
|
router, tenants, cleanup := newTestRouterSetup(t, 2)
|
||||||
|
defer cleanup()
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
pool, err := router.Resolve(ctx, tenants[0].Slug)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("resolve: %v", err)
|
||||||
|
}
|
||||||
|
var dbName string
|
||||||
|
if err := pool.QueryRow(ctx, `SELECT current_database()`).Scan(&dbName); err != nil {
|
||||||
|
t.Fatalf("current_database: %v", err)
|
||||||
|
}
|
||||||
|
if dbName != tenants[0].DBName {
|
||||||
|
t.Fatalf("current_database() = %q, want %q", dbName, tenants[0].DBName)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Akzeptanzkriterium 3 + Pruefung 2: fehlender/unbekannter Tenant-Kontext
|
||||||
|
// wird explizit abgewiesen statt irgendeine Verbindung zu liefern.
|
||||||
|
func TestRouter_RejectsMissingOrUnknownTenant(t *testing.T) {
|
||||||
|
router, _, cleanup := newTestRouterSetup(t, 1)
|
||||||
|
defer cleanup()
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
if _, err := router.Resolve(ctx, ""); !errors.Is(err, ErrMissingTenantContext) {
|
||||||
|
t.Fatalf("erwartet ErrMissingTenantContext, habe %v", err)
|
||||||
|
}
|
||||||
|
if _, err := router.Resolve(ctx, "nie-registrierter-slug"); !errors.Is(err, ErrUnknownTenant) {
|
||||||
|
t.Fatalf("erwartet ErrUnknownTenant, habe %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Akzeptanzkriterium 2 + Pruefung 3: Verbindungswiederverwendung nachweislich
|
||||||
|
// gemessen — zweiter Resolve-Aufruf liefert exakt denselben Pool, kein
|
||||||
|
// erneuter Verbindungsaufbau.
|
||||||
|
func TestRouter_ReusesConnectionForSameTenant(t *testing.T) {
|
||||||
|
router, tenants, cleanup := newTestRouterSetup(t, 1)
|
||||||
|
defer cleanup()
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
first, err := router.Resolve(ctx, tenants[0].Slug)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("resolve 1: %v", err)
|
||||||
|
}
|
||||||
|
second, err := router.Resolve(ctx, tenants[0].Slug)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("resolve 2: %v", err)
|
||||||
|
}
|
||||||
|
if first != second {
|
||||||
|
t.Fatal("erwartet identische pool-instanz bei wiederholtem resolve, habe unterschiedliche")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Akzeptanzkriterium 2 + Pruefung 1: Lasttest mit mehr simulierten Mandanten
|
||||||
|
// als maxOpen — die Zahl gleichzeitig offener Tenant-Pools bleibt begrenzt
|
||||||
|
// (LRU-Eviction), waechst also NICHT linear mit der Mandantenzahl.
|
||||||
|
func TestRouter_BoundsOpenConnectionsUnderLoad(t *testing.T) {
|
||||||
|
const tenantCount = 6
|
||||||
|
router, tenants, cleanup := newTestRouterSetup(t, tenantCount)
|
||||||
|
defer cleanup()
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
for _, tn := range tenants {
|
||||||
|
if _, err := router.Resolve(ctx, tn.Slug); err != nil {
|
||||||
|
t.Fatalf("resolve %s: %v", tn.Slug, err)
|
||||||
|
}
|
||||||
|
if router.OpenCount() > 2 {
|
||||||
|
t.Fatalf("OpenCount() = %d, erwartet <= maxOpen (2) nach jedem Resolve", router.OpenCount())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if router.OpenCount() != 2 {
|
||||||
|
t.Fatalf("erwartet genau maxOpen=2 offene pools nach %d tenants, habe %d", tenantCount, router.OpenCount())
|
||||||
|
}
|
||||||
|
|
||||||
|
// Evictete Tenants sind wieder ganz normal ueber die Registry aufloesbar
|
||||||
|
// (Cache-Miss fuehrt zu neuem, funktionierendem Pool, kein Fehlerzustand).
|
||||||
|
pool, err := router.Resolve(ctx, tenants[0].Slug)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("resolve nach eviction: %v", err)
|
||||||
|
}
|
||||||
|
var one int
|
||||||
|
if err := pool.QueryRow(ctx, `SELECT 1`).Scan(&one); err != nil {
|
||||||
|
t.Fatalf("query nach re-resolve: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1 +0,0 @@
|
|||||||
DROP TABLE IF EXISTS feature_flags;
|
|
||||||
@@ -1,10 +0,0 @@
|
|||||||
-- Feature-Flags zentral je Mandant/Zielgruppe (LIC-02, siehe core-kanban/tickets/LIC-02.md).
|
|
||||||
-- Lebt in der Registry-DB, nicht pro Tenant-Datenbank — Flags sind eine
|
|
||||||
-- Core-weite Konfiguration, keine Mandanten-Geschaeftsdaten.
|
|
||||||
CREATE TABLE feature_flags (
|
|
||||||
key TEXT PRIMARY KEY,
|
|
||||||
enabled BOOLEAN NOT NULL DEFAULT false,
|
|
||||||
rollout_percentage INT NOT NULL DEFAULT 0 CHECK (rollout_percentage BETWEEN 0 AND 100),
|
|
||||||
target_tenant_slugs TEXT[] NOT NULL DEFAULT '{}',
|
|
||||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
|
||||||
);
|
|
||||||
@@ -1,23 +0,0 @@
|
|||||||
#!/usr/bin/env bash
|
|
||||||
# Setzt die nexarch-Testumgebung zurueck: loescht die geteilte
|
|
||||||
# Registry-Tabelle "tenants" in der postgres-Wartungsdatenbank sowie alle
|
|
||||||
# tenant_*-Datenbanken. Noetig, weil verschiedene Feature-Branches
|
|
||||||
# unterschiedliche Registry-Schemata erwarten, aber dieselbe physische
|
|
||||||
# Postgres-Instanz auf dem Testhost teilen (siehe [[project-nexarch-test-infra]]).
|
|
||||||
#
|
|
||||||
# Aufruf: NEXARCH_TEST_DB_PASSWORD=... ./scripts/reset-test-env.sh
|
|
||||||
set -euo pipefail
|
|
||||||
|
|
||||||
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
|
|
||||||
ROLE="nexarch_test"
|
|
||||||
|
|
||||||
export PGPASSWORD="$PASS"
|
|
||||||
|
|
||||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS tenants CASCADE;"
|
|
||||||
|
|
||||||
dbs=$(psql -h localhost -U "$ROLE" -d postgres -tAc "SELECT datname FROM pg_database WHERE datname LIKE 'tenant\_%' ESCAPE '\'")
|
|
||||||
for db in $dbs; do
|
|
||||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP DATABASE IF EXISTS \"${db}\";"
|
|
||||||
done
|
|
||||||
|
|
||||||
echo "Testumgebung zurueckgesetzt: registry-tabelle + $(echo "$dbs" | grep -c . || true) tenant-datenbank(en) entfernt."
|
|
||||||
@@ -1,24 +0,0 @@
|
|||||||
#!/usr/bin/env bash
|
|
||||||
# Ein-Kommando-Pruefung fuer den aktuellen Code-Stand auf dem Testhost:
|
|
||||||
# Registry+Tenant-DBs zuruecksetzen, dann build/vet/test in einem Rutsch.
|
|
||||||
# -p 1 ist Pflicht, da mehrere Pakete dieselbe physische Registry-Tabelle auf
|
|
||||||
# dem Testhost teilen (siehe [[project-nexarch-test-infra]]).
|
|
||||||
#
|
|
||||||
# Aufruf: NEXARCH_TEST_DB_PASSWORD=... ./scripts/run-checks.sh
|
|
||||||
set -euo pipefail
|
|
||||||
|
|
||||||
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
|
|
||||||
cd "$(dirname "$0")/.."
|
|
||||||
|
|
||||||
NEXARCH_TEST_DB_PASSWORD="$PASS" bash scripts/reset-test-env.sh
|
|
||||||
|
|
||||||
export TEST_ADMIN_DSN="postgresql://nexarch_test:${PASS}@localhost:5432/postgres?sslmode=disable"
|
|
||||||
|
|
||||||
echo "== go build =="
|
|
||||||
go build ./...
|
|
||||||
|
|
||||||
echo "== go vet =="
|
|
||||||
go vet ./...
|
|
||||||
|
|
||||||
echo "== go test (-p 1) =="
|
|
||||||
go test ./... -p 1 -count=1
|
|
||||||
Reference in New Issue
Block a user