diff --git a/dms/cmd/app/api.go b/dms/cmd/app/api.go index 628e85e4..e56f945f 100644 --- a/dms/cmd/app/api.go +++ b/dms/cmd/app/api.go @@ -15,6 +15,7 @@ import ( "gitea.perlbach24.de/scripte/nexarch/dms/internal/apiwiring" "gitea.perlbach24.de/scripte/nexarch/dms/internal/cmis" dmscrypto "gitea.perlbach24.de/scripte/nexarch/dms/internal/crypto" + "gitea.perlbach24.de/scripte/nexarch/dms/internal/dbadmission" "gitea.perlbach24.de/scripte/nexarch/dms/internal/einvoice" "gitea.perlbach24.de/scripte/nexarch/dms/internal/folderapi" "gitea.perlbach24.de/scripte/nexarch/dms/internal/jobqueue" @@ -122,6 +123,18 @@ func envInt(name string, def int) int { return def } +// admissionGate baut das Zulassungs-Gate (FDN-28) aus der Umgebung. +// NEXARCH_DMS_APP_ADMISSION_MAX_CONCURRENT=0 schaltet die Kontrolle ab +// (Deps.Admission bleibt nil, Verhalten wie vor diesem Ticket). +func admissionGate() *dbadmission.Gate { + max := envInt("NEXARCH_DMS_APP_ADMISSION_MAX_CONCURRENT", 40) + if os.Getenv("NEXARCH_DMS_APP_ADMISSION_MAX_CONCURRENT") == "0" { + return nil + } + wait := envSeconds("NEXARCH_DMS_APP_ADMISSION_MAX_WAIT_SECONDS", 2*time.Second) + return dbadmission.NewGate(max, wait) +} + func envBool(name string, def bool) bool { v := strings.TrimSpace(strings.ToLower(os.Getenv(name))) if v == "" { @@ -244,7 +257,11 @@ func identityChain(next http.Handler, prefix string) (http.Handler, error) { if jwksURL == "" && !allowCaller { return nil, fmt.Errorf("keine identitaetsquelle: NEXARCH_DMS_APP_CORE_JWKS_URL setzen (oder uebergangsweise NEXARCH_DMS_APP_ALLOW_CALLER_IDENTITY=true)") } - cfg := apiwiring.ChainConfig{Prefix: prefix, AllowCaller: allowCaller, HSTSMaxAge: envSeconds("NEXARCH_DMS_APP_HSTS_MAX_AGE_SECONDS", 0)} + cfg := apiwiring.ChainConfig{ + Prefix: prefix, AllowCaller: allowCaller, HSTSMaxAge: envSeconds("NEXARCH_DMS_APP_HSTS_MAX_AGE_SECONDS", 0), + Admission: admissionGate(), + AdmissionRetryAfter: envSeconds("NEXARCH_DMS_APP_ADMISSION_RETRY_AFTER_SECONDS", 5*time.Second), + } if policyURL := os.Getenv("NEXARCH_DMS_APP_POLICY_URL"); policyURL != "" { token := os.Getenv("NEXARCH_DMS_APP_POLICY_SERVICE_TOKEN") if token == "" { diff --git a/dms/cmd/worker/main.go b/dms/cmd/worker/main.go index 8cee1ef2..453474cc 100644 --- a/dms/cmd/worker/main.go +++ b/dms/cmd/worker/main.go @@ -18,6 +18,7 @@ import ( "github.com/jackc/pgx/v5/pgxpool" dmscrypto "gitea.perlbach24.de/scripte/nexarch/dms/internal/crypto" + "gitea.perlbach24.de/scripte/nexarch/dms/internal/dbadmission" "gitea.perlbach24.de/scripte/nexarch/dms/internal/einvoice" "gitea.perlbach24.de/scripte/nexarch/dms/internal/errortracking" "gitea.perlbach24.de/scripte/nexarch/dms/internal/httpharden" @@ -34,8 +35,8 @@ import ( "gitea.perlbach24.de/scripte/nexarch/dms/internal/shared" "gitea.perlbach24.de/scripte/nexarch/dms/internal/storage" "gitea.perlbach24.de/scripte/nexarch/dms/internal/upload" - "gitea.perlbach24.de/scripte/nexarch/dms/internal/workflowevent" "gitea.perlbach24.de/scripte/nexarch/dms/internal/workerruntime" + "gitea.perlbach24.de/scripte/nexarch/dms/internal/workflowevent" ) func main() { @@ -262,14 +263,28 @@ func startJobWorkers(ctx context.Context, pool *pgxpool.Pool, queue *jobqueue.Qu log.Print("dms-worker: keine Workflow-Anbindung konfiguriert (NEXARCH_DMS_WORKER_WORKFLOW_URL), document.* wird nicht registriert/publiziert") } rt := workerruntime.New(deps) + admissionMax := 40 + if v, err := strconv.Atoi(os.Getenv("NEXARCH_DMS_WORKER_ADMISSION_MAX_CONCURRENT")); err == nil && v > 0 { + admissionMax = v + } + var gate *dbadmission.Gate + if os.Getenv("NEXARCH_DMS_WORKER_ADMISSION_MAX_CONCURRENT") != "0" { + admissionWait := 2 * time.Second + if v, err := strconv.Atoi(os.Getenv("NEXARCH_DMS_WORKER_ADMISSION_MAX_WAIT_SECONDS")); err == nil && v > 0 { + admissionWait = time.Duration(v) * time.Second + } + gate = dbadmission.NewGate(admissionMax, admissionWait) + } + admit := dbadmission.JobMiddleware(gate) rt.Middleware = func(next jobqueue.Handler) jobqueue.Handler { - return func(ctx context.Context, job *jobqueue.Job) error { + tracked := func(ctx context.Context, job *jobqueue.Job) error { err := next(ctx, job) if err != nil { tracker.Capture(ctx, "worker."+job.JobType, err) } return err } + return admit(tracked) } for _, n := range rt.Notices { log.Print("dms-worker: ", n) diff --git a/dms/docs/FDN-28-PRUEFPROTOKOLL.md b/dms/docs/FDN-28-PRUEFPROTOKOLL.md new file mode 100644 index 00000000..f42df5a8 --- /dev/null +++ b/dms/docs/FDN-28-PRUEFPROTOKOLL.md @@ -0,0 +1,132 @@ +# FDN-28 – Prüfprotokoll: Rückstau/Drosselung vor dem DB-Verbindungspool + +Nachtrag zu QA-10. QA-10 (`feature/qa-10-last-leistungstest`, Commit +`da34550`, noch nicht auf `main` gemerged) hatte bei 150 gleichzeitigen +Anfragen einen echten Engpass gefunden: Postgres meldete `too many clients +already` (SQLSTATE 53300), sowohl für HTTP-Anfragen als auch für +Hintergrundjobs (Malware-Scan, Such-Sync) — als rohe Verbindungsfehler statt +einer definierten `503`-Antwort. Dieses Ticket fügt die fehlende +Zulassungskontrolle (Admission Control) zwischen Anfragevolumen und +DB-Zugriff ein. + +**Hinweis zur Ausgangslage:** Der Ticket-Text nennt `dms/docs/QA-10-PRUEFPROTOKOLL.md` +als "jetzt auf main gemerged". Das stimmt zum Zeitpunkt dieser Arbeit nicht +— QA-10 liegt noch unverändert auf seinem eigenen Feature-Branch +(`feature/qa-10-last-leistungstest`), nicht auf `main`. Das Dokument wurde +trotzdem direkt aus diesem Branch gelesen (Abschnitt 3 dort beschreibt genau +den hier behobenen Engpass); die Implementierung dieses Tickets ist davon +unabhängig und baut auf dem aktuellen `main` auf. + +## 1. Lösung + +Neues Paket `dms/internal/dbadmission`: + +- `Gate`: Zulassungssemaphor (gepuffertes Kanal-Semaphor) mit konfigurierbarer + maximaler Nebenläufigkeit und maximaler Wartezeit auf einen freien + Durchlass. +- `Middleware(g, retryAfter)`: HTTP-Middleware — wird das Gate innerhalb der + Wartezeit nicht frei, antwortet sie mit `503` (Core-Fehlerformat, + `apierror.Write`, Code `unavailable`) und `Retry-After`-Header statt den + Request an den DB-zugreifenden Handler durchzulassen. +- `JobMiddleware(g)`: Job-Queue-Middleware — bei erschöpftem Gate wird ein + gewöhnlicher (transienter) Fehler zurückgegeben, kein + `jobqueue.LimitExceededError`. Das sorgt dafür, dass `Queue.Fail` den Job + mit exponentiellem Backoff erneut einplant (verzögerte Verarbeitung), + statt ihn in die Dead-Letter-Queue zu verschieben oder Daten zu verlieren. +- Ein `nil`-Gate lässt jede Anfrage/jeden Job unverändert durch (Admission + deaktiviert, z. B. für Bestandsinstallationen ohne explizite + Konfiguration). + +Verdrahtung: + +- `dms/internal/apiwiring.ChainConfig` bekommt `Admission *dbadmission.Gate` + und `AdmissionRetryAfter time.Duration`; `Chain()` schaltet die Middleware + vor Authentifizierung/Rechteentscheidung (frühestmöglicher Ausstieg bei + Überlast, bevor unnötige Arbeit anfällt). +- `dms/cmd/app/api.go`: `identityChain` baut das Gate aus der Umgebung + (`NEXARCH_DMS_APP_ADMISSION_MAX_CONCURRENT`, Default 40; + `NEXARCH_DMS_APP_ADMISSION_MAX_WAIT_SECONDS`, Default 2; + `NEXARCH_DMS_APP_ADMISSION_RETRY_AFTER_SECONDS`, Default 5; + `...MAX_CONCURRENT=0` schaltet ab). +- `dms/cmd/worker/main.go`: `rt.Middleware` wird um `dbadmission.JobMiddleware` + erweitert (eigene Env-Variablen `NEXARCH_DMS_WORKER_ADMISSION_MAX_CONCURRENT`/ + `..._MAX_WAIT_SECONDS`, gleiche Defaults/Abschaltlogik). Das Gate sitzt + außerhalb der bestehenden Tracker-Middleware, sodass Gate-Erschöpfung + selbst nicht als Fehler im Error-Tracking auftaucht (erwartetes + Rückstau-Verhalten, kein Bug). + +## 2. Automatisierte Tests + +``` +cd dms && go build ./... # OK +cd dms && go vet ./... # OK +cd dms && gofmt -l internal/dbadmission internal/apiwiring cmd/app cmd/worker # keine neuen Abweichungen +cd dms && TEST_TENANT_DSN=... TEST_MANTICORE_ADDR=127.0.0.1:9306 \ + go test ./internal/apiwiring/... ./internal/dbadmission/... \ + ./internal/jobqueue/... ./internal/workerruntime/... -p 1 -v +``` + +Ergebnis: alle Pakete **PASS**, keine Regression in bestehenden Tests. + +Neue Tests je Akzeptanzkriterium: + +| Test | Akzeptanzkriterium | Ergebnis | +|---|---|---| +| `dbadmission.TestMiddleware_UeberGrenze_503MitRetryAfter` | 1 | **bestanden** — 503 + `Retry-After` statt rohem Fehler | +| `apiwiring.TestChain_Admission_UeberGrenze_503MitRetryAfter` | 1 (reale Verdrahtung über `Chain()`, nicht nur isoliertes Paket) | **bestanden** | +| `dbadmission.TestJobMiddleware_ErschoepftesGate_VerzoegertStattDatenverlust` | 2 | **bestanden** — transienter Fehler statt `LimitExceededError`; Job a (vorher gestartet) läuft unverändert zu Ende durch | +| `dbadmission.TestMiddleware_UnterGrenze_Unveraendert` | 3 | **bestanden** — 5 Anfragen unterhalb der Grenze, alle 200 | +| `apiwiring.TestChain_OhneAdmission_Unveraendert` | 3 | **bestanden** — ohne konfiguriertes Gate (`ChainConfig{}`) unverändertes Verhalten | +| `dbadmission.TestGate_NilGate_LaesstAllesDurch` | 3 (Regression bei deaktivierter Kontrolle) | **bestanden** | + +## 3. Prüfungen vor der Abnahme + +**1. Simulierte Überlast:** `TestChain_Admission_UeberGrenze_503MitRetryAfter` +fährt die tatsächliche Produktions-`Chain()`-Funktion (nicht nur das +isolierte `dbadmission`-Paket) mit einem Gate der Kapazität 1: eine +blockierende Anfrage hält den einzigen Durchlass, eine zweite, gleichzeitige +Anfrage erhält `503` mit `Retry-After: 9` statt eines rohen +Verbindungsfehlers. **Bestanden.** + +Bewusst NICHT real gegen den geteilten Postgres-Dienst von `dms_dev` +reproduziert: Das würde (wie in QA-10 dokumentiert) echte +`max_connections`-Erschöpfung auf einem von mehreren Sitzungen gemeinsam +genutzten Testhost erfordern und andere parallele Arbeit auf 204 gefährden. +Die Zulassungskontrolle selbst ist DB-agnostisch (sie begrenzt Nebenläufigkeit +vor dem Handler-Aufruf, unabhängig davon, was der Handler tut) — der +HTTP-Ebenen-Test deckt den Mechanismus vollständig ab, ohne dieses Risiko +einzugehen. + +**2. Hintergrundjobs unter Überlast:** +`TestJobMiddleware_ErschoepftesGate_VerzoegertStattDatenverlust` zeigt: bei +erschöpftem Gate kommt ein transienter Fehler zurück (kein +`LimitExceededError`), der in der echten Job-Verarbeitung +(`jobqueue.Worker.processOne`) zu `Queue.Fail` (exponentielles Backoff, bis +zu `MaxAttempts` Versuche) statt zu einem sofortigen Fehlschlag/Datenverlust +führt. Der zuerst gestartete Job läuft unverändert bis zum Erfolg durch — +kein Job geht verloren. **Bestanden.** + +**3. Bestehende QA-10-Prüfungen laufen unverändert grün:** QA-10 ist nicht +auf `main` gemerged (siehe Hinweis oben) und daher auf diesem Branch nicht +vorhanden/ausführbar. Stattdessen wurde die volle bestehende Testsuite der +betroffenen Pakete (`apiwiring`, `jobqueue`, `workerruntime`) gegen die +reale `dms_dev`-Datenbank auf Host 204 gefahren — alle **PASS**, keine +Regression durch die neue Zulassungskontrolle. + +## 4. Gesamtergebnis + +| Akzeptanzkriterium | Status | +|---|---| +| 1. 503 + Retry-After statt rohem Verbindungsfehler bei Überlast | **erfüllt** | +| 2. Hintergrundjobs verzögert statt fehlzuschlagen | **erfüllt** | +| 3. Verhalten unterhalb der Grenze unverändert | **erfüllt** | + +| Prüfung vor der Abnahme | Status | +|---|---| +| 1. Simulierte Überlast: definierte 503-Antworten | **erfüllt** (HTTP-Ebene real gegen `Chain()`; keine reale Postgres-Verbindungserschöpfung auf geteiltem Testhost provoziert, siehe Begründung oben) | +| 2. Hintergrundjobs unter Überlast: verzögert, kein Datenverlust | **erfüllt** | +| 3. Bestehende QA-10-Prüfungen laufen unverändert grün | **nicht anwendbar** (QA-10 nicht auf `main`), stattdessen volle betroffene Testsuite real gegen `dms_dev` gefahren, grün | + +Alle drei Akzeptanzkriterien sind erfüllt. Prüfung 3 der Vorgabe war wegen +des nicht gemergten QA-10-Standes nicht wörtlich ausführbar; als Ersatz +wurde die reale, betroffene Testsuite gegen die Live-Datenbank gefahren. diff --git a/dms/internal/apiwiring/admission_chain_test.go b/dms/internal/apiwiring/admission_chain_test.go new file mode 100644 index 00000000..0d29a45e --- /dev/null +++ b/dms/internal/apiwiring/admission_chain_test.go @@ -0,0 +1,62 @@ +package apiwiring + +import ( + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "gitea.perlbach24.de/scripte/nexarch/dms/internal/dbadmission" +) + +// TestChain_Admission_UeberGrenze_503MitRetryAfter deckt FDN-28 +// Akzeptanzkriterium 1 auf Ebene von Chain (die tatsaechliche Verdrahtung +// vor den HTTP-Handlern, nicht nur das dbadmission-Paket isoliert). +func TestChain_Admission_UeberGrenze_503MitRetryAfter(t *testing.T) { + gate := dbadmission.NewGate(1, 20*time.Millisecond) + cfg := ChainConfig{Admission: gate, AdmissionRetryAfter: 9 * time.Second} + + release := make(chan struct{}) + started := make(chan struct{}) + handler := Chain(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(started) + <-release + w.WriteHeader(http.StatusOK) + }), cfg) + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/x", nil)) + }() + <-started + + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/x", nil)) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("status=%d, erwartet 503", rec.Code) + } + if got := rec.Header().Get("Retry-After"); got != "9" { + t.Fatalf("Retry-After=%q, erwartet 9", got) + } + + close(release) + wg.Wait() +} + +// TestChain_OhneAdmission_Unveraendert deckt FDN-28 Akzeptanzkriterium 3: +// ohne konfiguriertes Gate (wie vor diesem Ticket) bleibt das Verhalten +// unveraendert. +func TestChain_OhneAdmission_Unveraendert(t *testing.T) { + handler := Chain(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }), ChainConfig{}) + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/x", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("status=%d, erwartet 200", rec.Code) + } +} diff --git a/dms/internal/apiwiring/apiwiring.go b/dms/internal/apiwiring/apiwiring.go index 90f70e0e..2c92dc8b 100644 --- a/dms/internal/apiwiring/apiwiring.go +++ b/dms/internal/apiwiring/apiwiring.go @@ -15,6 +15,7 @@ import ( "gitea.perlbach24.de/scripte/nexarch/dms/internal/bulkedit" "gitea.perlbach24.de/scripte/nexarch/dms/internal/bulkeditapi" dmscrypto "gitea.perlbach24.de/scripte/nexarch/dms/internal/crypto" + "gitea.perlbach24.de/scripte/nexarch/dms/internal/dbadmission" "gitea.perlbach24.de/scripte/nexarch/dms/internal/documentapi" "gitea.perlbach24.de/scripte/nexarch/dms/internal/downloadapi" "gitea.perlbach24.de/scripte/nexarch/dms/internal/einvoice" @@ -123,10 +124,17 @@ type ChainConfig struct { Policy *policyclient.EnforceConfig // HSTSMaxAge > 0 setzt Strict-Transport-Security (nur hinter TLS sinnvoll). HSTSMaxAge time.Duration + // Admission begrenzt gleichzeitige Anfragen vor dem DB-Pool (FDN-28); + // nil = keine Zulassungskontrolle (Verhalten wie vor diesem Ticket). + Admission *dbadmission.Gate + // AdmissionRetryAfter ist der Retry-After-Wert bei 503 durch Admission + // (<=0 = dbadmission.DefaultRetryAfter). + AdmissionRetryAfter time.Duration } -// Chain schaltet vor next: einheitliches Fehlerformat, Authentifizierung -// (Core-JWT) und, wenn konfiguriert, die zentrale Rechteentscheidung. +// Chain schaltet vor next: Zulassungskontrolle vor dem DB-Pool, einheitliches +// Fehlerformat, Authentifizierung (Core-JWT) und, wenn konfiguriert, die +// zentrale Rechteentscheidung. func Chain(next http.Handler, c ChainConfig) http.Handler { handler := next if c.Policy != nil { @@ -137,5 +145,8 @@ func Chain(next http.Handler, c ChainConfig) http.Handler { Verifier: c.Verifier, Prefix: c.Prefix, AllowCaller: c.AllowCaller, Public: openapi.PublicPaths(), }, handler) } + if c.Admission != nil { + handler = dbadmission.Middleware(c.Admission, c.AdmissionRetryAfter)(handler) + } return SecurityHeaders(apierror.Normalize(handler), c.HSTSMaxAge) } diff --git a/dms/internal/dbadmission/dbadmission.go b/dms/internal/dbadmission/dbadmission.go new file mode 100644 index 00000000..257fd8eb --- /dev/null +++ b/dms/internal/dbadmission/dbadmission.go @@ -0,0 +1,110 @@ +// Package dbadmission begrenzt gleichzeitige Anfragen/Jobs vor dem +// Datenbank-Verbindungspool (FDN-28, Nachtrag zu QA-10): statt roher +// Postgres-Verbindungsfehler (SQLSTATE 53300) bei Ueberlast liefert HTTP +// eine definierte 503-Antwort mit Retry-After, Hintergrundjobs werden +// verzoegert (Backoff ueber jobqueue.Queue.Fail) statt mit Verbindungsfehler +// abzubrechen. +package dbadmission + +import ( + "context" + "errors" + "net/http" + "strconv" + "time" + + "gitea.perlbach24.de/scripte/nexarch/dms/internal/apierror" + "gitea.perlbach24.de/scripte/nexarch/dms/internal/jobqueue" +) + +// Gate ist ein Zulassungssemaphor mit begrenzter Wartezeit vor dem +// eigentlichen DB-Zugriff. Ein nil-*Gate laesst jede Anfrage/jeden Job +// unveraendert durch (Admission deaktiviert). +type Gate struct { + slots chan struct{} + maxWait time.Duration +} + +// NewGate baut ein Gate mit maxConcurrent gleichzeitigen Durchlaessen; +// maxWait ist die Zeit, die eine Anfrage auf einen freien Durchlass wartet, +// bevor sie als ausgeschoepft gilt. +func NewGate(maxConcurrent int, maxWait time.Duration) *Gate { + if maxConcurrent <= 0 { + maxConcurrent = 1 + } + return &Gate{slots: make(chan struct{}, maxConcurrent), maxWait: maxWait} +} + +func (g *Gate) acquire(ctx context.Context) (release func(), ok bool) { + if g == nil { + return func() {}, true + } + select { + case g.slots <- struct{}{}: + return func() { <-g.slots }, true + default: + } + timer := time.NewTimer(g.maxWait) + defer timer.Stop() + select { + case g.slots <- struct{}{}: + return func() { <-g.slots }, true + case <-ctx.Done(): + return nil, false + case <-timer.C: + return nil, false + } +} + +// DefaultRetryAfter ist der Fallback-Wert fuer den Retry-After-Header, wenn +// RetryAfter nicht gesetzt (<=0) ist. +const DefaultRetryAfter = 5 * time.Second + +// Middleware schaltet vor next eine Zulassungskontrolle: wird das Gate +// innerhalb von maxWait nicht frei, antwortet die API mit 503 und +// Retry-After statt den rohen DB-Verbindungsfehler zu erreichen. +func Middleware(g *Gate, retryAfter time.Duration) func(http.Handler) http.Handler { + if retryAfter <= 0 { + retryAfter = DefaultRetryAfter + } + return func(next http.Handler) http.Handler { + if g == nil { + return next + } + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + release, ok := g.acquire(r.Context()) + if !ok { + w.Header().Set("Retry-After", strconv.Itoa(int(retryAfter.Seconds()))) + apierror.Write(w, http.StatusServiceUnavailable, "unavailable", "dienst ueberlastet, bitte spaeter erneut versuchen") + return + } + defer release() + next.ServeHTTP(w, r) + }) + } +} + +// errGateExhausted ist bewusst ein gewoehnlicher (transienter) Fehler statt +// jobqueue.LimitExceededError: so greift Queue.Fail mit Backoff/Wiederholung +// statt den Job sofort in die Dead-Letter-Queue zu verschieben (FDN-28 +// Akzeptanzkriterium 2 – Rueckstau statt Datenverlust). +var errGateExhausted = errors.New("dbadmission: zulassungsgrenze erreicht, job wird spaeter erneut versucht") + +// JobMiddleware verzoegert Hintergrundjobs (Malware-Scan, Such-Sync, ...) bei +// erschoepftem Gate statt sie mit einem rohen DB-Verbindungsfehler scheitern +// zu lassen. +func JobMiddleware(g *Gate) func(jobqueue.Handler) jobqueue.Handler { + return func(next jobqueue.Handler) jobqueue.Handler { + if g == nil { + return next + } + return func(ctx context.Context, job *jobqueue.Job) error { + release, ok := g.acquire(ctx) + if !ok { + return errGateExhausted + } + defer release() + return next(ctx, job) + } + } +} diff --git a/dms/internal/dbadmission/dbadmission_test.go b/dms/internal/dbadmission/dbadmission_test.go new file mode 100644 index 00000000..5da457f4 --- /dev/null +++ b/dms/internal/dbadmission/dbadmission_test.go @@ -0,0 +1,128 @@ +package dbadmission + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "gitea.perlbach24.de/scripte/nexarch/dms/internal/jobqueue" +) + +// TestMiddleware_UnterGrenze_Unveraendert deckt Akzeptanzkriterium 3: unterhalb +// der Zulassungsgrenze bleibt das Verhalten unveraendert (kein 503, keine +// Verzoegerung). +func TestMiddleware_UnterGrenze_Unveraendert(t *testing.T) { + g := NewGate(2, 50*time.Millisecond) + handler := Middleware(g, time.Second)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + + for i := 0; i < 5; i++ { + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/x", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("anfrage %d: status=%d, erwartet 200", i, rec.Code) + } + } +} + +// TestMiddleware_UeberGrenze_503MitRetryAfter deckt Akzeptanzkriterium 1: bei +// Ueberschreiten der Zulassungsgrenze antwortet die API mit 503 und +// Retry-After statt eines rohen Verbindungsfehlers. +func TestMiddleware_UeberGrenze_503MitRetryAfter(t *testing.T) { + g := NewGate(1, 30*time.Millisecond) + release := make(chan struct{}) + started := make(chan struct{}) + slow := Middleware(g, 7*time.Second)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(started) + <-release + w.WriteHeader(http.StatusOK) + })) + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + rec := httptest.NewRecorder() + slow.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/x", nil)) + }() + <-started + + rec := httptest.NewRecorder() + slow.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/x", nil)) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("status=%d, erwartet 503", rec.Code) + } + if got := rec.Header().Get("Retry-After"); got != "7" { + t.Fatalf("Retry-After=%q, erwartet 7", got) + } + if ct := rec.Header().Get("Content-Type"); ct != "application/json" { + t.Fatalf("Content-Type=%q, erwartet application/json (Core-Fehlerformat)", ct) + } + + close(release) + wg.Wait() +} + +// TestJobMiddleware_ErschoepftesGate_VerzoegertStattDatenverlust deckt +// Akzeptanzkriterium 2: Hintergrundjobs laufen bei Ueberlast verzoegert +// (transienter Fehler -> Queue.Fail-Backoff) statt mit einem rohen +// Verbindungsfehler zu scheitern; LimitExceededError (-> Dead-Letter) wird +// dabei nicht ausgeloest. +func TestJobMiddleware_ErschoepftesGate_VerzoegertStattDatenverlust(t *testing.T) { + g := NewGate(1, 20*time.Millisecond) + release := make(chan struct{}) + started := make(chan struct{}) + slow := JobMiddleware(g)(func(ctx context.Context, job *jobqueue.Job) error { + close(started) + <-release + return nil + }) + + done := make(chan error, 1) + go func() { done <- slow(context.Background(), &jobqueue.Job{ID: "a"}) }() + <-started + + err := slow(context.Background(), &jobqueue.Job{ID: "b"}) + if err == nil { + t.Fatal("erwarteter Fehler bei erschoepftem Gate blieb aus") + } + var limitErr *jobqueue.LimitExceededError + if errors.As(err, &limitErr) { + t.Fatal("erschoepftes Gate darf keinen LimitExceededError (-> Dead-Letter) ausloesen, sondern einen transienten Fehler (-> Backoff)") + } + + close(release) + if err := <-done; err != nil { + t.Fatalf("job a: unerwarteter fehler: %v", err) + } +} + +// TestGate_NilGate_LaesstAllesDurch stellt sicher, dass ein abgeschaltetes +// Gate (nil) weder HTTP noch Jobs beeinflusst. +func TestGate_NilGate_LaesstAllesDurch(t *testing.T) { + handler := Middleware(nil, time.Second)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/x", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("status=%d, erwartet 200", rec.Code) + } + + called := false + wrapped := JobMiddleware(nil)(func(ctx context.Context, job *jobqueue.Job) error { + called = true + return nil + }) + if err := wrapped(context.Background(), &jobqueue.Job{}); err != nil { + t.Fatal(err) + } + if !called { + t.Fatal("job-handler wurde bei nil-gate nicht aufgerufen") + } +}