diff --git a/ai/docs/GOV-04-PRUEFPROTOKOLL.md b/ai/docs/GOV-04-PRUEFPROTOKOLL.md new file mode 100644 index 00000000..d4b60cc8 --- /dev/null +++ b/ai/docs/GOV-04-PRUEFPROTOKOLL.md @@ -0,0 +1,64 @@ +# GOV-04 — Prüfprotokoll: Kostenkontrolle & Quota (Kopplung Core LIC-03) + +Datum: 2026-09-27 +Testhost: root@192.168.1.204 (Wegwerf-Rolle/DB `gov04_test`, nach Abschluss entfernt) + +## Ausgangslage geprüft + +- `ENG-05` (`ai/internal/queue`): Postgres-Warteschlange für Inferenz-Aufrufe — vorausgesetzt, unverändert im Kernverhalten. +- Core `LIC-03` (`internal/usage`): Nutzungszähler/Quotas je Tenant, HTTP-Kopplung über `internal/resync.Handler.UsageHandler` (Delta-Meldung, "applied genau einmal"). AI ist ein eigenständiges Go-Modul (`ai/go.mod`) und kann `internal/usage` nicht importieren — Muster identisch zu `dms/internal/storage.HTTPUsageReporter` / `mail/internal/storage`: lokaler Zähler/Quota-Spiegel + Delta-Meldung an Core. +- `GOV-03` (`ai/internal/residency`): Sibling-Muster für tenant-gescopte Governance-Konfiguration in diesem Modul (Store + strukturelle Interfaces statt harter Kopplung). + +## Umsetzung + +- `ai/internal/quota`: `Store` mit `Check`/`Enforce`/`RecordCall`, Metrik `ai_inference_calls`, Warnschwelle 80 % (analog Core `internal/usage`), `HTTPReporter` meldet jedes Delta an Cores Resync-Endpunkt (Kopplung LIC-03). +- `ai/migrations/0007_ai_usage_quota.{up,down}.sql`: lokale Tabellen `ai_usage_counters`/`ai_usage_quotas`. +- `ai/migrations/0008_inference_jobs_tenant.{up,down}.sql`: `tenant_slug`-Spalte auf `inference_jobs` (nullable), damit ENG-05-Jobs einem Tenant zuordenbar sind. +- `ai/internal/queue`: `QuotaEnforcer`-Interface, `Queue.SetQuotaEnforcer` — `Enqueue` blockt VOR dem Anlegen eines Jobs, wenn die Tenant-Quota erreicht ist (AK2). +- `ai/internal/queue`: `UsageRecorder`-Interface, `Worker.SetUsageRecorder` — Zählung erst NACH erfolgreichem `Complete`, sodass der Zählerstand tatsächlich ausgeführte Aufrufe abbildet (Prüfung 3), nicht bloße Versuche. + +## Akzeptanzkriterien + +| # | Kriterium | Ergebnis | +|---|---|---| +| 1 | Verbrauch ist je Tenant nachvollziehbar erfasst | Erfüllt — `quota.Store.Check`/`Get`, strikt `tenant_slug`-gescoped, siehe `TestStore_CheckTracksPerTenant`, `TestQuota_TenantIsolation` | +| 2 | Überschreitung der Quota wird zuverlässig verhindert | Erfüllt — `Queue.Enqueue` blockt vor Jobanlage, siehe `TestStore_EnforceBlocksWhenExceeded`, `TestQuota_BlocksEnqueueAndTracksActualCompletions` | +| 3 | Näherung an die Grenze wird vorab gemeldet | Erfüllt — `Store.Enforce` löst bei 80 % `WarnFunc` aus, VOR Erreichen der Grenze, siehe `TestStore_EnforceWarnsBeforeExceeding` | + +## Prüfungen vor der Abnahme + +### 1. Quota-Überschreitung im Test zuverlässig blockiert + +Zielwert: bei erreichter Tenant-Quota wird kein weiterer Job angelegt, `Enqueue` liefert `quota.ErrQuotaExceeded`. + +Ergebnis: **bestanden**. `TestStore_EnforceBlocksWhenExceeded` (reine Store-Logik) und `TestQuota_BlocksEnqueueAndTracksActualCompletions` (End-to-End über echte Queue + Worker + Postgres): nach 3 tatsächlich abgeschlossenen Aufrufen bei Limit 3 wird der 4. `Enqueue`-Versuch abgelehnt, `Queue.Depth().Pending` bleibt bei 0 — kein Job entsteht. + +### 2. Vorabwarnung im Test ausgelöst + +Zielwert: bei 80 % Auslastung wird die Warnreaktion ausgelöst, der Aufruf selbst bleibt aber noch erlaubt. + +Ergebnis: **bestanden**. `TestStore_EnforceWarnsBeforeExceeding`: bei 8 von 10 konfigurierten Aufrufen (80 %) löst `Enforce` genau eine Warnung aus und lässt den Aufruf weiterhin zu. + +### 3. Verbrauchszählung gegen tatsächliche Aufrufe abgeglichen + +Zielwert: der Zählerstand entspricht der Anzahl tatsächlich über die Queue ausgeführter (abgeschlossener) Inferenz-Aufrufe, nicht der Anzahl Enqueue-Versuche. + +Ergebnis: **bestanden**. `TestQuota_BlocksEnqueueAndTracksActualCompletions`: 3 Jobs werden über `Worker.ProcessOne` real mit einem Mock-Provider ausgeführt (Dequeue → execute → Complete); `quota.Store.Get` liefert danach exakt 3 — die Zählung erfolgt im Worker erst nach erfolgreichem `Complete`, ein abgelehnter/blockierter Enqueue-Versuch erhöht den Zähler nicht. Die Kopplung an Core LIC-03 (Delta-Meldung je erfassten Aufruf) ist zusätzlich durch `TestStore_RecordCallReportsToCore` (Fake-Reporter erfasst genau die gemeldeten Deltas) nachgewiesen. + +## Testausführung + +``` +TEST_TENANT_DSN=postgres://gov04_test:***@localhost:5432/gov04_test?sslmode=disable \ + go test ./ai/internal/quota/... ./ai/internal/queue/... -v +``` + +Alle 14 Tests (5 `quota`, 9 `queue`, davon 4 neu für GOV-04) **PASS**, inklusive der bereits bestehenden ENG-05/ENG-07-Tests (keine Regression). + +`go build ./...` und `go vet ./...` im `ai`-Modul: keine Fehler. + +Voller Modul-Testlauf `go test ./...`: alle Pakete bis auf `ai/internal/vectorindex` grün; `TestSearch_LatencyAtScale` in `vectorindex` läuft auf diesem Testhost in einen 10-Minuten-Timeout — vorbestehend, unabhängig von GOV-04 (kein Bezug zu Queue/Quota-Code), nicht Teil dieser Kachel und daher nicht behoben. + +## Offene Punkte / Entscheidungen + +- Core exponiert (Stand LIC-03) nur einen Write-Endpunkt (`resync.UsageHandler`, Delta anwenden) — keinen synchronen Read/Check-Endpunkt für Module. Enforcement erfolgt daher lokal gegen einen AI-eigenen Zähler-/Quota-Spiegel (identisches Muster zu dms/mail), Core wird informativ nachgeführt. Das ist eine reversible, kleine Design-Entscheidung im Rahmen der bestehenden Konventionen, keine Abweichung von den Akzeptanzkriterien. +- Setzen der Quota-Obergrenze je Tenant (Admin-Oberfläche) ist Core `LIC-04` und damit ausdrücklich nicht Teil dieser Kachel; `quota.Store.SetQuota` existiert als Schnittstelle dafür. diff --git a/ai/internal/queue/queue.go b/ai/internal/queue/queue.go index 17fb5d9d..9a77d5a5 100644 --- a/ai/internal/queue/queue.go +++ b/ai/internal/queue/queue.go @@ -47,14 +47,24 @@ type Job struct { MaxAttempts int Result json.RawMessage LastError *string + TenantSlug string } const DefaultMaxAttempts = 5 +// QuotaEnforcer ist die schmale Schnittstelle zu GOV-04 (ai/internal/quota): +// blockiert Enqueue, wenn die Tenant-Quota bereits erreicht ist. Als +// Interface hier statt Import, damit queue (ENG-05) nicht von quota +// abhaengt — quota.Store erfuellt diese Schnittstelle strukturell. +type QuotaEnforcer interface { + Enforce(ctx context.Context, tenantSlug string) error +} + type Queue struct { pool *pgxpool.Pool staleLockAfter time.Duration limits limits.Limits + quota QuotaEnforcer } func NewQueue(pool *pgxpool.Pool, staleLockAfter time.Duration) *Queue { @@ -65,14 +75,22 @@ func NewQueue(pool *pgxpool.Pool, staleLockAfter time.Duration) *Queue { // vor allem fuer Tests mit kleinen, deterministischen Werten. func (q *Queue) SetLimits(l limits.Limits) { q.limits = l } +// SetQuotaEnforcer verknuepft die Queue mit GOV-04s Kostenkontrolle; ohne +// Aufruf bleibt Enqueue unveraendert unbegrenzt (Rueckwaertskompatibilitaet). +func (q *Queue) SetQuotaEnforcer(e QuotaEnforcer) { q.quota = e } + type EnqueueOptions struct { IdempotencyKey string MaxAttempts int + TenantSlug string } // Enqueue reiht einen Inferenz-Aufruf asynchron ein (AK1). Bei gesetztem // IdempotencyKey liefert ein erneuter Aufruf mit demselben Key die ID des -// bereits vorhandenen Jobs statt eines Duplikats. +// bereits vorhandenen Jobs statt eines Duplikats. Ist ein QuotaEnforcer +// gesetzt und TenantSlug angegeben, wird die Tenant-Quota VOR dem Anlegen +// des Jobs geprueft (GOV-04 AK2) — eine ueberschrittene Quota verhindert +// zuverlaessig, dass ueberhaupt erst neue Arbeit entsteht. func (q *Queue) Enqueue(ctx context.Context, jobType Kind, payload any, opts EnqueueOptions) (string, error) { payloadJSON, err := json.Marshal(payload) if err != nil { @@ -81,6 +99,11 @@ func (q *Queue) Enqueue(ctx context.Context, jobType Kind, payload any, opts Enq if err := q.limits.CheckSize(len(payloadJSON)); err != nil { return "", err } + if q.quota != nil && opts.TenantSlug != "" { + if err := q.quota.Enforce(ctx, opts.TenantSlug); err != nil { + return "", err + } + } maxAttempts := opts.MaxAttempts if maxAttempts <= 0 { maxAttempts = DefaultMaxAttempts @@ -89,13 +112,17 @@ func (q *Queue) Enqueue(ctx context.Context, jobType Kind, payload any, opts Enq if opts.IdempotencyKey != "" { idempotencyKey = opts.IdempotencyKey } + var tenantSlug any + if opts.TenantSlug != "" { + tenantSlug = opts.TenantSlug + } var id string err = q.pool.QueryRow(ctx, ` - INSERT INTO inference_jobs (job_type, payload, idempotency_key, max_attempts) - VALUES ($1, $2, $3, $4) + INSERT INTO inference_jobs (job_type, payload, idempotency_key, max_attempts, tenant_slug) + VALUES ($1, $2, $3, $4, $5) ON CONFLICT (idempotency_key) DO UPDATE SET job_type = inference_jobs.job_type RETURNING id - `, string(jobType), payloadJSON, idempotencyKey, maxAttempts).Scan(&id) + `, string(jobType), payloadJSON, idempotencyKey, maxAttempts, tenantSlug).Scan(&id) if err != nil { return "", fmt.Errorf("queue: job einreihen: %w", err) } @@ -112,7 +139,7 @@ func (q *Queue) Dequeue(ctx context.Context, workerID string) (*Job, error) { defer func() { _ = tx.Rollback(ctx) }() row := tx.QueryRow(ctx, ` - SELECT id, job_type, payload, status, attempts, max_attempts, last_error + SELECT id, job_type, payload, status, attempts, max_attempts, last_error, COALESCE(tenant_slug, '') FROM inference_jobs WHERE ( (status = 'pending' AND available_at <= now()) @@ -125,7 +152,7 @@ func (q *Queue) Dequeue(ctx context.Context, workerID string) (*Job, error) { var j Job var jobType string - if err := row.Scan(&j.ID, &jobType, &j.Payload, &j.Status, &j.Attempts, &j.MaxAttempts, &j.LastError); err != nil { + if err := row.Scan(&j.ID, &jobType, &j.Payload, &j.Status, &j.Attempts, &j.MaxAttempts, &j.LastError, &j.TenantSlug); err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, ErrNoJobAvailable } @@ -193,9 +220,9 @@ func (q *Queue) Status(ctx context.Context, jobID string) (*Job, error) { var j Job var jobType string err := q.pool.QueryRow(ctx, ` - SELECT id, job_type, payload, status, attempts, max_attempts, result, last_error + SELECT id, job_type, payload, status, attempts, max_attempts, result, last_error, COALESCE(tenant_slug, '') FROM inference_jobs WHERE id = $1 - `, jobID).Scan(&j.ID, &jobType, &j.Payload, &j.Status, &j.Attempts, &j.MaxAttempts, &j.Result, &j.LastError) + `, jobID).Scan(&j.ID, &jobType, &j.Payload, &j.Status, &j.Attempts, &j.MaxAttempts, &j.Result, &j.LastError, &j.TenantSlug) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, ErrNotFound diff --git a/ai/internal/queue/quota_integration_test.go b/ai/internal/queue/quota_integration_test.go new file mode 100644 index 00000000..6b7a8f51 --- /dev/null +++ b/ai/internal/queue/quota_integration_test.go @@ -0,0 +1,99 @@ +package queue + +import ( + "context" + "errors" + "testing" + "time" + + "gitea.perlbach24.de/scripte/nexarch/ai/internal/provider" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/provider/mock" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/quota" +) + +// GOV-04 Pruefung 1 + 3: eine reale, ueber die Queue (ENG-05) ausgefuehrte +// Inferenz-Last erhoeht den Tenant-Zaehler exakt um die Anzahl tatsaechlich +// abgeschlossener Aufrufe, und eine erreichte Quota verhindert zuverlaessig, +// dass weitere Jobs ueberhaupt erst eingereiht werden. +func TestQuota_BlocksEnqueueAndTracksActualCompletions(t *testing.T) { + pool := newPool(t) + quotaStore := quota.NewStore(pool, nil) + ctx := context.Background() + + const tenant = "tenant-integration" + const limit = 3 + if err := quotaStore.SetQuota(ctx, tenant, quota.InferenceCallsMetric, limit); err != nil { + t.Fatalf("quota setzen: %v", err) + } + + q := NewQueue(pool, time.Minute) + q.SetQuotaEnforcer(quotaStore) + + p := &mock.Provider{ChatResponse: provider.ChatResponse{Message: provider.Message{Content: "ok"}}} + w := NewWorker(q, p, "worker-quota-test") + w.SetUsageRecorder(quotaStore) + + // Genau `limit` Auftraege einreihen und abarbeiten -> danach ist die + // Quota erreicht (AK1: Zaehlerstand nachvollziehbar je Tenant). + for i := 0; i < limit; i++ { + if _, err := q.Enqueue(ctx, KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{TenantSlug: tenant}); err != nil { + t.Fatalf("enqueue %d haette erlaubt sein muessen: %v", i, err) + } + if err := w.ProcessOne(ctx); err != nil { + t.Fatalf("verarbeitung %d fehlgeschlagen: %v", i, err) + } + } + + value, err := quotaStore.Get(ctx, tenant, quota.InferenceCallsMetric) + if err != nil { + t.Fatalf("zaehlerstand lesen: %v", err) + } + if value != limit { + t.Fatalf("erwartet zaehlerstand %d (= tatsaechlich ausgefuehrte aufrufe), bekommen %d", limit, value) + } + + // AK2 / Pruefung 1: der naechste Enqueue-Versuch wird zuverlaessig + // abgelehnt, BEVOR ein weiterer Job entsteht. + _, err = q.Enqueue(ctx, KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{TenantSlug: tenant}) + if !errors.Is(err, quota.ErrQuotaExceeded) { + t.Fatalf("enqueue nach erreichter quota haette blockiert werden muessen, bekommen: %v", err) + } + + depth, err := q.Depth(ctx) + if err != nil { + t.Fatalf("warteschlangenstand lesen: %v", err) + } + if depth.Pending != 0 { + t.Fatalf("blockierter enqueue-versuch haette keinen job anlegen duerfen, pending=%d", depth.Pending) + } +} + +// Ein Tenant ohne konfigurierte Quota bleibt unbeeinflusst von einem Tenant, +// der seine Quota bereits erreicht hat (kein globaler, sondern strikt +// tenant-gescopter Zustand). +func TestQuota_TenantIsolation(t *testing.T) { + pool := newPool(t) + quotaStore := quota.NewStore(pool, nil) + ctx := context.Background() + + if err := quotaStore.SetQuota(ctx, "tenant-blocked", quota.InferenceCallsMetric, 1); err != nil { + t.Fatalf("quota setzen: %v", err) + } + + q := NewQueue(pool, time.Minute) + q.SetQuotaEnforcer(quotaStore) + + if _, err := q.Enqueue(ctx, KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{TenantSlug: "tenant-blocked"}); err != nil { + t.Fatalf("erster enqueue haette erlaubt sein muessen: %v", err) + } + if err := quotaStore.RecordCall(ctx, "tenant-blocked"); err != nil { + t.Fatalf("call erfassen: %v", err) + } + + if _, err := q.Enqueue(ctx, KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{TenantSlug: "tenant-blocked"}); !errors.Is(err, quota.ErrQuotaExceeded) { + t.Fatalf("tenant-blocked haette blockiert sein muessen: %v", err) + } + if _, err := q.Enqueue(ctx, KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{TenantSlug: "tenant-unaffected"}); err != nil { + t.Fatalf("ein anderer tenant ohne quota-treffer sollte unbeeinflusst bleiben: %v", err) + } +} diff --git a/ai/internal/queue/worker.go b/ai/internal/queue/worker.go index f506192d..8d681e12 100644 --- a/ai/internal/queue/worker.go +++ b/ai/internal/queue/worker.go @@ -12,6 +12,14 @@ import ( "gitea.perlbach24.de/scripte/nexarch/ai/internal/provider" ) +// UsageRecorder ist die schmale Schnittstelle zu GOV-04 (ai/internal/quota): +// zaehlt einen tatsaechlich ausgefuehrten Inferenz-Aufruf. Als Interface +// hier statt Import, damit queue (ENG-05) nicht von quota abhaengt — +// quota.Store erfuellt diese Schnittstelle strukturell. +type UsageRecorder interface { + RecordCall(ctx context.Context, tenantSlug string) error +} + // Worker holt Jobs aus der Queue und fuehrt sie ueber den Provider aus. // Ein Ausfuehrungsfehler blockiert den Anfrage-Pfad nicht (AK1) — der // aufrufende Code hat laengst eine Job-ID zurueckbekommen und pollt oder @@ -21,6 +29,7 @@ type Worker struct { provider provider.Provider id string limits limits.Limits + usage UsageRecorder } func NewWorker(q *Queue, p provider.Provider, workerID string) *Worker { @@ -31,6 +40,10 @@ func NewWorker(q *Queue, p provider.Provider, workerID string) *Worker { // vor allem fuer Tests mit kleinen, deterministischen Werten. func (w *Worker) SetLimits(l limits.Limits) { w.limits = l } +// SetUsageRecorder verknuepft den Worker mit GOV-04s Verbrauchszaehlung; ohne +// Aufruf bleibt die Ausfuehrung unveraendert ungezaehlt. +func (w *Worker) SetUsageRecorder(u UsageRecorder) { w.usage = u } + // ProcessOne holt einen Job und fuehrt ihn aus. Liefert ErrNoJobAvailable, // wenn die Queue leer ist (kein Fehlerzustand). func (w *Worker) ProcessOne(ctx context.Context) error { @@ -54,6 +67,16 @@ func (w *Worker) ProcessOne(ctx context.Context) error { if completeErr := w.queue.Complete(ctx, job.ID, result); completeErr != nil { return fmt.Errorf("queue: job %s abschliessen: %w", job.ID, completeErr) } + // Zaehlung erst NACH erfolgreichem Abschluss (GOV-04 Pruefung 3): der + // Zaehlerstand bildet tatsaechlich ausgefuehrte Aufrufe ab, nicht + // blosse Versuche — ein Fehlschlag der Zaehlung faellt bereits + // erfolgreiche Inferenz nicht rueckwirkend zurueck, sondern wird nur + // geloggt. + if w.usage != nil && job.TenantSlug != "" { + if usageErr := w.usage.RecordCall(ctx, job.TenantSlug); usageErr != nil { + log.Printf("queue: verbrauchszaehlung fuer job %s (tenant %s) fehlgeschlagen: %v", job.ID, job.TenantSlug, usageErr) + } + } return nil } diff --git a/ai/internal/quota/quota.go b/ai/internal/quota/quota.go new file mode 100644 index 00000000..da9eae8d --- /dev/null +++ b/ai/internal/quota/quota.go @@ -0,0 +1,183 @@ +// Package quota implementiert GOV-04: Verbrauchszaehlung und Obergrenzen fuer +// KI-Inferenz je Tenant, gekoppelt an Core LIC-03 (Nutzungszaehler & Quotas). +// AI ist ein eigenes Go-Modul (ai/go.mod) und kann Core internal/usage nicht +// importieren (siehe ai/internal/migrate-Paketkommentar) — Store fuehrt +// dieselbe Zaehler-/Quota-Semantik lokal und meldet jedes Delta zusaetzlich +// an Core weiter (Reporter), damit ein Tenant genau einen konsistenten +// Nutzungsstand fuer Cores Verwaltungsoberflaeche hat (dms/mail-Muster, +// siehe dms/internal/storage.HTTPUsageReporter). +package quota + +import ( + "context" + "errors" + "fmt" + "log/slog" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// InferenceCallsMetric ist der Metrikname fuer gezaehlte KI-Inferenz-Aufrufe. +const InferenceCallsMetric = "ai_inference_calls" + +// ErrQuotaExceeded blockiert einen neuen Inferenz-Aufruf (Akzeptanzkriterium 2). +var ErrQuotaExceeded = errors.New("quota: ki-inferenz-quota fuer diesen tenant ueberschritten") + +// Status ist die aus Zaehlerstand/Quota abgeleitete Reaktion. +type Status string + +const ( + StatusOK Status = "ok" + StatusWarning Status = "warning" // Schwelle (80%) erreicht, noch nicht ueberschritten + StatusExceeded Status = "exceeded" // Quota erreicht/ueberschritten +) + +const warningThreshold = 0.8 + +// Reporter meldet ein angewendetes Delta an Core LIC-03 weiter. Schmale +// Schnittstelle, damit Tests einen Fake statt eines echten HTTP-Aufrufs +// einsetzen koennen. +type Reporter interface { + Report(ctx context.Context, tenantSlug, metric string, delta int64) error +} + +// WarnFunc wird bei StatusWarning aufgerufen (Akzeptanzkriterium 3), bevor +// die Quota ueberschritten wird — Standard ist ein strukturiertes Log. +type WarnFunc func(tenantSlug string, value, limit int64) + +type Store struct { + pool *pgxpool.Pool + reporter Reporter + warn WarnFunc +} + +// NewStore verknuepft den lokalen Zaehler mit einem optionalen Reporter an +// Core LIC-03 (reporter darf nil sein, z.B. in Tests ohne Core-Anbindung). +func NewStore(pool *pgxpool.Pool, reporter Reporter) *Store { + return &Store{pool: pool, reporter: reporter, warn: defaultWarn} +} + +func defaultWarn(tenantSlug string, value, limit int64) { + slog.Warn("ki-inferenz-quota naehert sich der grenze", "tenant", tenantSlug, "value", value, "limit", limit) +} + +// SetWarnFunc ueberschreibt die Warnreaktion (Tests, andere Benachrichtigungskanaele). +func (s *Store) SetWarnFunc(f WarnFunc) { s.warn = f } + +func (s *Store) increment(ctx context.Context, tenantSlug, metric string, delta int64) error { + _, err := s.pool.Exec(ctx, ` + INSERT INTO ai_usage_counters (tenant_slug, metric, value, updated_at) + VALUES ($1, $2, $3, now()) + ON CONFLICT (tenant_slug, metric) DO UPDATE + SET value = ai_usage_counters.value + $3, updated_at = now() + `, tenantSlug, metric, delta) + if err != nil { + return fmt.Errorf("quota: zaehler erhoehen: %w", err) + } + return nil +} + +// Get liefert den aktuellen Zaehlerstand — 0, wenn noch nie erhoeht wurde, +// strikt tenant-gescoped (Akzeptanzkriterium 1). +func (s *Store) Get(ctx context.Context, tenantSlug, metric string) (int64, error) { + var value int64 + err := s.pool.QueryRow(ctx, ` + SELECT value FROM ai_usage_counters WHERE tenant_slug = $1 AND metric = $2 + `, tenantSlug, metric).Scan(&value) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return 0, nil + } + return 0, fmt.Errorf("quota: zaehler lesen: %w", err) + } + return value, nil +} + +// SetQuota legt die Obergrenze fuer (tenantSlug, metric) fest — Konfiguration, +// kein Hardcode (Produkt-DNA: Datenresidenz-/Quota-Grenzen konfigurierbar). +func (s *Store) SetQuota(ctx context.Context, tenantSlug, metric string, limit int64) error { + _, err := s.pool.Exec(ctx, ` + INSERT INTO ai_usage_quotas (tenant_slug, metric, limit_value) + VALUES ($1, $2, $3) + ON CONFLICT (tenant_slug, metric) DO UPDATE SET limit_value = $3 + `, tenantSlug, metric, limit) + if err != nil { + return fmt.Errorf("quota: obergrenze setzen: %w", err) + } + return nil +} + +func (s *Store) getQuota(ctx context.Context, tenantSlug, metric string) (limit int64, configured bool, err error) { + err = s.pool.QueryRow(ctx, ` + SELECT limit_value FROM ai_usage_quotas WHERE tenant_slug = $1 AND metric = $2 + `, tenantSlug, metric).Scan(&limit) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return 0, false, nil + } + return 0, false, fmt.Errorf("quota: obergrenze lesen: %w", err) + } + return limit, true, nil +} + +// Check liefert Zaehlerstand, konfigurierte Obergrenze und den daraus +// abgeleiteten Status (Akzeptanzkriterium 1). Ohne konfigurierte Obergrenze +// gilt die Metrik als unbegrenzt (StatusOK). +func (s *Store) Check(ctx context.Context, tenantSlug, metric string) (value, limit int64, status Status, err error) { + value, err = s.Get(ctx, tenantSlug, metric) + if err != nil { + return 0, 0, "", err + } + limit, configured, err := s.getQuota(ctx, tenantSlug, metric) + if err != nil { + return 0, 0, "", err + } + if !configured { + return value, 0, StatusOK, nil + } + switch { + case value >= limit: + return value, limit, StatusExceeded, nil + case limit > 0 && float64(value) >= warningThreshold*float64(limit): + return value, limit, StatusWarning, nil + default: + return value, limit, StatusOK, nil + } +} + +// Enforce blockiert einen neuen Inferenz-Aufruf, wenn die Quota fuer +// tenantSlug bereits erreicht ist (Akzeptanzkriterium 2), und loest vorab +// die Warnreaktion aus, wenn sich der Tenant der Grenze naehert +// (Akzeptanzkriterium 3) — beides VOR dem eigentlichen Aufruf. +func (s *Store) Enforce(ctx context.Context, tenantSlug string) error { + value, limit, status, err := s.Check(ctx, tenantSlug, InferenceCallsMetric) + if err != nil { + return fmt.Errorf("quota: pruefung fehlgeschlagen: %w", err) + } + if status == StatusExceeded { + return fmt.Errorf("%w (tenant %s)", ErrQuotaExceeded, tenantSlug) + } + if status == StatusWarning && s.warn != nil { + s.warn(tenantSlug, value, limit) + } + return nil +} + +// RecordCall zaehlt genau einen tatsaechlich ausgefuehrten Inferenz-Aufruf +// (Akzeptanzkriterium 1) und meldet das Delta an Core LIC-03 weiter. Der +// Aufrufer ruft RecordCall NUR nach erfolgreicher Ausfuehrung auf, damit der +// Zaehlerstand mit tatsaechlichen Aufrufen uebereinstimmt (Pruefung 3), nicht +// mit blossen Versuchen. Eine fehlgeschlagene Core-Meldung wird geloggt, +// blockiert aber nicht den bereits erfolgreich abgeschlossenen Aufruf. +func (s *Store) RecordCall(ctx context.Context, tenantSlug string) error { + if err := s.increment(ctx, tenantSlug, InferenceCallsMetric, 1); err != nil { + return err + } + if s.reporter != nil { + if err := s.reporter.Report(ctx, tenantSlug, InferenceCallsMetric, 1); err != nil { + slog.Error("quota: meldung an core lic-03 fehlgeschlagen", "tenant", tenantSlug, "error", err) + } + } + return nil +} diff --git a/ai/internal/quota/quota_test.go b/ai/internal/quota/quota_test.go new file mode 100644 index 00000000..88c5b1e5 --- /dev/null +++ b/ai/internal/quota/quota_test.go @@ -0,0 +1,184 @@ +package quota + +import ( + "context" + "errors" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/ai/internal/migrate" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/testdb" +) + +func newPool(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := os.Getenv("TEST_TENANT_DSN") + if dsn == "" { + t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen") + } + pool := testdb.Pool(t, dsn) + + root, err := filepath.Abs("../../migrations") + if err != nil { + t.Fatalf("migrationsverzeichnis: %v", err) + } + migrations, err := migrate.Load(root) + if err != nil { + t.Fatalf("migrationen laden: %v", err) + } + if _, err := migrate.Up(context.Background(), pool, migrations); err != nil { + t.Fatalf("migrationen anwenden: %v", err) + } + return pool +} + +// fakeReporter erfasst gemeldete Deltas, ohne einen echten HTTP-Aufruf an +// Core zu machen — beweist, dass RecordCall die Meldung tatsaechlich ausloest +// (Kopplung an Core LIC-03). +type fakeReporter struct { + mu sync.Mutex + reports []usageDeltaDTO +} + +func (f *fakeReporter) Report(_ context.Context, tenantSlug, metric string, delta int64) error { + f.mu.Lock() + defer f.mu.Unlock() + f.reports = append(f.reports, usageDeltaDTO{TenantSlug: tenantSlug, Metric: metric, Delta: delta}) + return nil +} + +// AK1: Verbrauch ist je Tenant nachvollziehbar erfasst. +func TestStore_CheckTracksPerTenant(t *testing.T) { + pool := newPool(t) + store := NewStore(pool, nil) + ctx := context.Background() + + if err := store.SetQuota(ctx, "tenant-a", InferenceCallsMetric, 10); err != nil { + t.Fatalf("quota setzen: %v", err) + } + if err := store.RecordCall(ctx, "tenant-a"); err != nil { + t.Fatalf("call erfassen: %v", err) + } + if err := store.RecordCall(ctx, "tenant-a"); err != nil { + t.Fatalf("call erfassen: %v", err) + } + + value, limit, status, err := store.Check(ctx, "tenant-a", InferenceCallsMetric) + if err != nil { + t.Fatalf("check: %v", err) + } + if value != 2 || limit != 10 || status != StatusOK { + t.Fatalf("erwartet value=2 limit=10 status=ok, bekommen value=%d limit=%d status=%s", value, limit, status) + } + + // Ein anderer Tenant bleibt unbeeinflusst. + otherValue, err := store.Get(ctx, "tenant-b", InferenceCallsMetric) + if err != nil { + t.Fatalf("get tenant-b: %v", err) + } + if otherValue != 0 { + t.Fatalf("tenant-b sollte unbeeinflusst sein, bekommen %d", otherValue) + } +} + +// AK2 / Pruefung 1: Quota-Ueberschreitung wird zuverlaessig verhindert. +func TestStore_EnforceBlocksWhenExceeded(t *testing.T) { + pool := newPool(t) + store := NewStore(pool, nil) + ctx := context.Background() + + if err := store.SetQuota(ctx, "tenant-exceed", InferenceCallsMetric, 2); err != nil { + t.Fatalf("quota setzen: %v", err) + } + for i := 0; i < 2; i++ { + if err := store.Enforce(ctx, "tenant-exceed"); err != nil { + t.Fatalf("call %d haette erlaubt sein muessen: %v", i, err) + } + if err := store.RecordCall(ctx, "tenant-exceed"); err != nil { + t.Fatalf("call %d erfassen: %v", i, err) + } + } + + err := store.Enforce(ctx, "tenant-exceed") + if !errors.Is(err, ErrQuotaExceeded) { + t.Fatalf("dritter aufruf haette blockiert werden muessen, bekommen: %v", err) + } +} + +// AK3 / Pruefung 2: Naeherung an die Grenze wird vorab gemeldet, BEVOR die +// Quota ueberschritten ist. +func TestStore_EnforceWarnsBeforeExceeding(t *testing.T) { + pool := newPool(t) + store := NewStore(pool, nil) + ctx := context.Background() + + var mu sync.Mutex + var warnings []string + store.SetWarnFunc(func(tenantSlug string, value, limit int64) { + mu.Lock() + defer mu.Unlock() + warnings = append(warnings, tenantSlug) + }) + + if err := store.SetQuota(ctx, "tenant-warn", InferenceCallsMetric, 10); err != nil { + t.Fatalf("quota setzen: %v", err) + } + // 8 von 10 = 80% -> Warnschwelle erreicht, aber noch erlaubt. + for i := 0; i < 8; i++ { + if err := store.RecordCall(ctx, "tenant-warn"); err != nil { + t.Fatalf("call %d erfassen: %v", i, err) + } + } + + if err := store.Enforce(ctx, "tenant-warn"); err != nil { + t.Fatalf("bei 80%% sollte der aufruf noch erlaubt sein: %v", err) + } + + mu.Lock() + got := len(warnings) + mu.Unlock() + if got != 1 { + t.Fatalf("erwartet genau eine warnung vor ueberschreitung, bekommen %d", got) + } +} + +// Pruefung 3: Verbrauchszaehlung wird an Core LIC-03 weitergemeldet (Kopplung). +func TestStore_RecordCallReportsToCore(t *testing.T) { + pool := newPool(t) + reporter := &fakeReporter{} + store := NewStore(pool, reporter) + ctx := context.Background() + + if err := store.RecordCall(ctx, "tenant-report"); err != nil { + t.Fatalf("call erfassen: %v", err) + } + if err := store.RecordCall(ctx, "tenant-report"); err != nil { + t.Fatalf("call erfassen: %v", err) + } + + reporter.mu.Lock() + defer reporter.mu.Unlock() + if len(reporter.reports) != 2 { + t.Fatalf("erwartet 2 meldungen an core, bekommen %d", len(reporter.reports)) + } + for _, r := range reporter.reports { + if r.TenantSlug != "tenant-report" || r.Metric != InferenceCallsMetric || r.Delta != 1 { + t.Fatalf("unerwartete meldung: %+v", r) + } + } +} + +// Ohne konfigurierte Quota gilt eine Metrik als unbegrenzt. +func TestStore_NoQuotaConfiguredMeansUnlimited(t *testing.T) { + pool := newPool(t) + store := NewStore(pool, nil) + ctx := context.Background() + + if err := store.Enforce(ctx, "tenant-unlimited"); err != nil { + t.Fatalf("ohne konfigurierte quota sollte enforce nie blockieren: %v", err) + } +} diff --git a/ai/internal/quota/reporter.go b/ai/internal/quota/reporter.go new file mode 100644 index 00000000..8506cc58 --- /dev/null +++ b/ai/internal/quota/reporter.go @@ -0,0 +1,62 @@ +package quota + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" +) + +// usageDeltaDTO entspricht Core internal/resync.usageDeltaDTO (JSON-Vertrag: +// tenant_slug/metric/delta) — hier gespiegelt wie im dms/mail-Muster +// (dms/internal/storage.HTTPUsageReporter), da AI Cores internal/-Pakete +// nicht importieren kann. +type usageDeltaDTO struct { + TenantSlug string `json:"tenant_slug"` + Metric string `json:"metric"` + Delta int64 `json:"delta"` +} + +// HTTPReporter meldet ueber Cores Resync-Nutzungs-Endpunkt +// (internal/resync.Handler.UsageHandler, LIC-03/API-06) — authentifiziert +// ueber dasselbe Service-Credential-Verfahren wie jeder andere Modul-Core- +// Aufruf (API-02). +type HTTPReporter struct { + endpointURL string + clientID string + clientSecret string + httpClient *http.Client +} + +func NewHTTPReporter(endpointURL, clientID, clientSecret string, httpClient *http.Client) *HTTPReporter { + if httpClient == nil { + httpClient = http.DefaultClient + } + return &HTTPReporter{endpointURL: endpointURL, clientID: clientID, clientSecret: clientSecret, httpClient: httpClient} +} + +func (r *HTTPReporter) Report(ctx context.Context, tenantSlug, metric string, delta int64) error { + body, err := json.Marshal([]usageDeltaDTO{{TenantSlug: tenantSlug, Metric: metric, Delta: delta}}) + if err != nil { + return fmt.Errorf("quota: nutzungsmeldung serialisieren: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.endpointURL, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("quota: nutzungsmeldungs-anfrage aufbauen: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Nexarch-Client-Id", r.clientID) + req.Header.Set("X-Nexarch-Client-Secret", r.clientSecret) + + resp, err := r.httpClient.Do(req) + if err != nil { + return fmt.Errorf("quota: nutzungsmeldung senden: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("quota: nutzungsmeldung von core abgelehnt: status %d", resp.StatusCode) + } + return nil +} diff --git a/ai/migrations/0007_ai_usage_quota.down.sql b/ai/migrations/0007_ai_usage_quota.down.sql new file mode 100644 index 00000000..0a60be73 --- /dev/null +++ b/ai/migrations/0007_ai_usage_quota.down.sql @@ -0,0 +1,2 @@ +DROP TABLE ai_usage_quotas; +DROP TABLE ai_usage_counters; diff --git a/ai/migrations/0007_ai_usage_quota.up.sql b/ai/migrations/0007_ai_usage_quota.up.sql new file mode 100644 index 00000000..9a7e4654 --- /dev/null +++ b/ai/migrations/0007_ai_usage_quota.up.sql @@ -0,0 +1,18 @@ +-- GOV-04: lokaler Nutzungszaehler/Quota-Spiegel je Tenant fuer KI-Inferenz. +-- AI ist ein eigenes Go-Modul und kann Core internal/usage (LIC-03) nicht +-- importieren; Struktur entspricht 1:1 Core migrations/0005_usage_counters, +-- Deltas werden zusaetzlich an Core gemeldet (siehe ai/internal/quota). +CREATE TABLE ai_usage_counters ( + tenant_slug TEXT NOT NULL, + metric TEXT NOT NULL, + value BIGINT NOT NULL DEFAULT 0, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (tenant_slug, metric) +); + +CREATE TABLE ai_usage_quotas ( + tenant_slug TEXT NOT NULL, + metric TEXT NOT NULL, + limit_value BIGINT NOT NULL, + PRIMARY KEY (tenant_slug, metric) +); diff --git a/ai/migrations/0008_inference_jobs_tenant.down.sql b/ai/migrations/0008_inference_jobs_tenant.down.sql new file mode 100644 index 00000000..6e55070e --- /dev/null +++ b/ai/migrations/0008_inference_jobs_tenant.down.sql @@ -0,0 +1 @@ +ALTER TABLE inference_jobs DROP COLUMN tenant_slug; diff --git a/ai/migrations/0008_inference_jobs_tenant.up.sql b/ai/migrations/0008_inference_jobs_tenant.up.sql new file mode 100644 index 00000000..9fe64d4e --- /dev/null +++ b/ai/migrations/0008_inference_jobs_tenant.up.sql @@ -0,0 +1,4 @@ +-- GOV-04: Tenant-Bezug je Inferenz-Job, damit Verbrauch/Quota (ai_usage_*) +-- ENG-05-Jobs zuverlaessig zuordnen kann. Nullable: bestehende Jobs ohne +-- Tenant-Angabe bleiben gueltig. +ALTER TABLE inference_jobs ADD COLUMN tenant_slug TEXT;