From 53d02bd4cfdcd38eba3103a195b11f8939b2d188 Mon Sep 17 00:00:00 2001 From: patrick Date: Sun, 27 Sep 2026 09:30:18 +0200 Subject: [PATCH] ENG-08: ressourcenlimits-rate-limiting-f-r-drittsystem-sync-jobs Co-Authored-By: Claude Sonnet 5 --- connect/docs/ENG-08-PRUEFPROTOKOLL.md | 36 +++++ connect/internal/syncjob/limits.go | 99 +++++++++++++ connect/internal/syncjob/limits_test.go | 105 ++++++++++++++ .../internal/thirdpartyrate/thirdpartyrate.go | 131 ++++++++++++++++++ .../thirdpartyrate/thirdpartyrate_test.go | 100 +++++++++++++ 5 files changed, 471 insertions(+) create mode 100644 connect/docs/ENG-08-PRUEFPROTOKOLL.md create mode 100644 connect/internal/syncjob/limits.go create mode 100644 connect/internal/syncjob/limits_test.go create mode 100644 connect/internal/thirdpartyrate/thirdpartyrate.go create mode 100644 connect/internal/thirdpartyrate/thirdpartyrate_test.go diff --git a/connect/docs/ENG-08-PRUEFPROTOKOLL.md b/connect/docs/ENG-08-PRUEFPROTOKOLL.md new file mode 100644 index 00000000..5e6ea37c --- /dev/null +++ b/connect/docs/ENG-08-PRUEFPROTOKOLL.md @@ -0,0 +1,36 @@ +# ENG-08 – Prüfprotokoll: Ressourcenlimits & Rate-Limiting für Drittsystem-Sync-Jobs + +Datum: 2026-09-27. Host: Entwicklungshost 192.168.1.204, Wegwerf-Cluster (Postgres über `TEST_TENANT_DSN`, per-Test-Schema via `connect/internal/testdb`). Klon in eigenem Wurfverzeichnis `/root/nexarch-forks/eng-08-connect`, synchronisiert per rsync aus lokalem Klon `feature/eng-08-ressourcenlimits-rate-limiting-f-r-drittsystem-sync-jobs`, `main`-Ausgangsstand `b8b8197e`. + +## Umsetzung + +Aufsetzend auf ENG-03 (`connect/internal/syncjob`, Sync-Job-Verwaltung) und ENG-05 (`syncjob.RetryPolicy`/`CompleteRunWithRetry`, dort bereits mit Abbruchgrenze `MaxAttempts` implementiert): + +- `connect/internal/syncjob/limits.go` (neu): `RunLimits{MaxDuration, MaxBytes}` und `RunWithLimits`, das die vom Aufrufer übergebene Sync-Arbeit unter einer Kontext-Deadline (`MaxDuration`) laufen lässt und ihr einen `ByteCounter` zum Melden übertragener Bytes gibt. Eine Grenzverletzung wird mit `syncjob.Permanent` markiert, sodass `CompleteRunWithRetry` den Run terminal beendet statt ihn — trotz evtl. verbleibender Retry-Versuche — endlos zu wiederholen. Konfiguration erfolgt wie bei `RetryPolicy` als vom Aufrufer übergebener Wert je Sync-Job/Connector-Typ (kein neues DB-Schema nötig). +- `connect/internal/thirdpartyrate` (neu): Token-Bucket-Ratenbegrenzer je Schlüssel (Connector-Instanz), plus `RoundTripper` zum transparenten Einhängen in den ausgehenden `*http.Client` eines Connectors (z. B. `erpconnector`/`crmconnector`). Jeder Schlüssel hat einen eigenen, unabhängigen Bucket — ein Connector, der sein Limit ausreizt, entzieht keinem anderen Schlüssel Kapazität. +- Keine Migrationen, keine neuen Zugangsdaten/Geheimnisse. + +## Prüfungen + +| # | Prüfung | Ziel | Ergebnis | +|---|---------|------|----------| +| 1 | Sync-Job überschreitet Laufzeit-/Datenmengenobergrenze → kontrollierter Abbruch | `RunWithLimits` beendet die Arbeit spätestens bei `MaxDuration`/`MaxBytes`, Fehler ist `ErrRunLimitExceeded` und `Permanent` | **Bestanden.** `TestRunWithLimits_MaxDurationExceeded_AbortsControlled` (Abbruch nach 50ms-Deadline) und `TestRunWithLimits_MaxBytesExceeded_AbortsControlled` (Abbruch bei 120 von 100 erlaubten Bytes) — beide `Permanent`. Gegenprobe `TestRunWithLimits_WithinLimits_Succeeds`: innerhalb der Grenzen kein Fehler. | +| 2 | Retry-Abbruchgrenze aus ENG-05 wird nach konfigurierter Anzahl Versuche durchgesetzt | Job endet nach `MaxAttempts` terminal, kein weiterer Retry | **Bestanden** (bereits durch ENG-05 abgedeckt, hier erneut real ausgeführt): `TestCompleteRunWithRetry_MaxAttemptsReachedEndsTerminally` — nach Erreichen von `MaxAttempts=2` liefert `ClaimDueRetries` 0 weitere Runs. Zusätzlich neu für ENG-08: `TestCompleteRunWithRetry_LimitExceededEndsTerminallyRegardlessOfAttemptsLeft` — ein Ressourcenlimit-Fehler beendet den Run terminal, obwohl `MaxAttempts=10` noch 9 Versuche übrig lässt; `ClaimDueRetries` liefert danach ebenfalls 0. | +| 3 | Lasttest: ein Connector mit Drittsystem-Fehlverhalten blockiert nicht die Sync-Jobs anderer Connectoren/Mandanten | Ein "böswilliger" Schlüssel, der ununterbrochen gegen ein sehr enges Limit (1 req/s) rennt, darf einen unabhängigen Schlüssel (1000 req/s) nicht verlangsamen | **Bestanden.** `TestWait_MisbehavingConnectorDoesNotBlockOtherConnector`: parallel zu einer Endlosschleife von `connector-bad` gegen sein 1-req/s-Limit absolviert `connector-good` 500 Anfragen in real < 1s (statt der 500s, die bei geteiltem Zustand nötig wären). Zusätzlich `TestWait_LimitsRequestsPerSecond` (Grundfunktion der Ratenbegrenzung selbst) und `TestRoundTripper_WaitsBeforeForwarding` (Einhängung in `*http.Client` real per `httptest.Server`). | + +## Entscheidungen + +- **In-Process statt Postgres-gestützter Ratenbegrenzer für AC3.** Core hat mit `internal/ratelimit` bereits einen Postgres-gestützten, prozessübergreifenden Rate-Limiter (API-03) — für Connect existiert aber noch kein Dispatcher-Prozess, der Sync-Runs tatsächlich mehrfach-instanziert ausführt (siehe dokumentierte Lücke in `masterdatasync.Runner.RunDueJobs`, ERP-03). Ein In-Process-Token-Bucket je Connector-Instanz ist die kleinste Lösung, die das Akzeptanzkriterium (Isolation zwischen Connectoren/Mandanten) erfüllt, ohne eine noch nicht vorhandene Mehrprozess-Ausführung vorwegzunehmen. Sobald ein echter Multi-Instanz-Dispatcher entsteht, sollte der Bucket-Zustand nach Postgres wandern (analog `internal/ratelimit`) — als Folgeticket vorzumerken, falls gewünscht. +- **Limits als Aufrufer-Parameter statt neue Tabelle.** Folgt demselben Muster wie `RetryPolicy` (ENG-05): "konfigurierbar" heißt hier, dass der Aufrufer (der künftige Sync-Dispatcher) die Werte je Job/Connector-Typ setzt, nicht dass sie per UI editierbar in einer neuen Tabelle liegen. Kein Schema-Zuwachs, kleinste Lösung. +- AC2 war durch ENG-05 bereits vollständig implementiert und getestet; für ENG-08 wurde die bestehende Prüfung real erneut ausgeführt (siehe Tabelle) und um einen Test ergänzt, der die neue Limits-Komponente in denselben Terminierungspfad einspeist. + +## Testlauf (real, Host 204) + +``` +go build ./... -> clean +go vet ./... -> clean +gofmt -l . (nur bereits vorher vorhandene, unveraenderte Dateien) -> keine neuen Funde +go test -count=1 ./internal/thirdpartyrate/... -v -> PASS (4/4) +go test -count=1 ./internal/syncjob/... -v -> PASS (18/18, inkl. neuer Limits-Tests) +go test -count=1 ./... -> alle Pakete ok +``` diff --git a/connect/internal/syncjob/limits.go b/connect/internal/syncjob/limits.go new file mode 100644 index 00000000..b486236e --- /dev/null +++ b/connect/internal/syncjob/limits.go @@ -0,0 +1,99 @@ +// ENG-08: hard per-run ceilings on runtime and transferred data, so a +// misconfigured or misbehaving connector cannot bind sync-queue resources +// indefinitely (the Zapier/n8n anti-pattern the ticket names). Enforced by +// RunWithLimits, which wraps the connector-specific work a caller runs for +// one sync_job_run. +// +// AC1: RunLimits.MaxDuration/MaxBytes are configurable per call (per +// connector/job), following the same pattern as RetryPolicy in retry.go. +// AC2 (durchgesetzt hier zusammen mit retry.go): a limit violation is +// Permanent, so CompleteRunWithRetry ends the run terminally instead of +// retrying an unbounded connector forever. +package syncjob + +import ( + "context" + "errors" + "fmt" + "sync" + "time" +) + +// RunLimits are the hard ceilings for a single sync_job_run. Zero means "no +// limit" for that dimension. +type RunLimits struct { + MaxDuration time.Duration + MaxBytes int64 +} + +// ErrRunLimitExceeded marks a run aborted for exceeding RunLimits. +var ErrRunLimitExceeded = errors.New("syncjob: run limit exceeded") + +// ByteCounter lets the wrapped work report transferred bytes as it streams +// data from/to the third system, so RunWithLimits can abort mid-transfer +// instead of only after the fact. +type ByteCounter struct { + mu sync.Mutex + limit int64 + total int64 + cancel context.CancelFunc +} + +// Add records n more transferred bytes and reports ErrRunLimitExceeded (and +// cancels the run's context) once the total exceeds the configured limit. +func (c *ByteCounter) Add(n int64) error { + if c == nil || c.limit <= 0 { + return nil + } + c.mu.Lock() + c.total += n + exceeded := c.total > c.limit + total := c.total + c.mu.Unlock() + if exceeded { + c.cancel() + return fmt.Errorf("%w: %d bytes uebertragen, limit %d", ErrRunLimitExceeded, total, c.limit) + } + return nil +} + +// Total returns the bytes recorded via Add so far. +func (c *ByteCounter) Total() int64 { + if c == nil { + return 0 + } + c.mu.Lock() + defer c.mu.Unlock() + return c.total +} + +// RunWithLimits runs work under limits.MaxDuration (enforced via a context +// deadline) and gives it a ByteCounter to enforce limits.MaxBytes. Either +// violation aborts work's context and returns a Permanent-wrapped +// ErrRunLimitExceeded, so a caller feeding the result into +// CompleteRunWithRetry gets a controlled, terminal end of the run (AC1/AC2) +// instead of an endless retry loop against a connector that is genuinely +// misbehaving. +func RunWithLimits(ctx context.Context, limits RunLimits, work func(ctx context.Context, bytes *ByteCounter) error) error { + runCtx := ctx + var cancel context.CancelFunc + if limits.MaxDuration > 0 { + runCtx, cancel = context.WithTimeout(ctx, limits.MaxDuration) + } else { + runCtx, cancel = context.WithCancel(ctx) + } + defer cancel() + + counter := &ByteCounter{limit: limits.MaxBytes, cancel: cancel} + err := work(runCtx, counter) + if err == nil { + return nil + } + if errors.Is(err, ErrRunLimitExceeded) { + return Permanent(err) + } + if errors.Is(runCtx.Err(), context.DeadlineExceeded) { + return Permanent(fmt.Errorf("%w: laufzeit ueber %s", ErrRunLimitExceeded, limits.MaxDuration)) + } + return err +} diff --git a/connect/internal/syncjob/limits_test.go b/connect/internal/syncjob/limits_test.go new file mode 100644 index 00000000..e16d6189 --- /dev/null +++ b/connect/internal/syncjob/limits_test.go @@ -0,0 +1,105 @@ +package syncjob + +import ( + "context" + "errors" + "testing" + "time" +) + +// AC1 / Pruefung 1: Laufzeitobergrenze wird durchgesetzt, Job endet +// kontrolliert statt endlos zu laufen. +func TestRunWithLimits_MaxDurationExceeded_AbortsControlled(t *testing.T) { + limits := RunLimits{MaxDuration: 50 * time.Millisecond} + err := RunWithLimits(context.Background(), limits, func(ctx context.Context, bytes *ByteCounter) error { + <-ctx.Done() + return ctx.Err() + }) + if err == nil { + t.Fatal("erwarteter Laufzeit-Fehler ausgeblieben") + } + if !errors.Is(err, ErrRunLimitExceeded) { + t.Fatalf("err = %v, erwartet ErrRunLimitExceeded", err) + } + if !IsPermanent(err) { + t.Fatal("Laufzeitueberschreitung muss Permanent sein (kein endloses Retry)") + } +} + +// AC1 / Pruefung 1: Datenmengenobergrenze wird durchgesetzt. +func TestRunWithLimits_MaxBytesExceeded_AbortsControlled(t *testing.T) { + limits := RunLimits{MaxBytes: 100} + err := RunWithLimits(context.Background(), limits, func(ctx context.Context, bytes *ByteCounter) error { + if addErr := bytes.Add(60); addErr != nil { + return addErr + } + if addErr := bytes.Add(60); addErr != nil { + return addErr + } + return nil + }) + if err == nil { + t.Fatal("erwarteter Datenmengen-Fehler ausgeblieben") + } + if !errors.Is(err, ErrRunLimitExceeded) { + t.Fatalf("err = %v, erwartet ErrRunLimitExceeded", err) + } + if !IsPermanent(err) { + t.Fatal("Datenmengenueberschreitung muss Permanent sein (kein endloses Retry)") + } +} + +func TestRunWithLimits_WithinLimits_Succeeds(t *testing.T) { + limits := RunLimits{MaxDuration: time.Second, MaxBytes: 1000} + err := RunWithLimits(context.Background(), limits, func(ctx context.Context, bytes *ByteCounter) error { + if addErr := bytes.Add(50); addErr != nil { + return addErr + } + return nil + }) + if err != nil { + t.Fatalf("unerwarteter Fehler innerhalb der Grenzen: %v", err) + } +} + +// AC2 / Pruefung 2: eine Grenzverletzung fliesst als Permanent-Fehler in +// CompleteRunWithRetry ein und beendet den Run terminal, unabhaengig von +// verbleibenden Retry-Versuchen (kein endloses Wiederholen eines +// Ressourcen-DoS). +func TestCompleteRunWithRetry_LimitExceededEndsTerminallyRegardlessOfAttemptsLeft(t *testing.T) { + s := newStore(t) + ctx := context.Background() + jobID, err := s.Create(ctx, Ref{TenantSlug: "t1", ConnectorName: "erp", ConnectorInstanceID: "i1"}, 60) + if err != nil { + t.Fatalf("create: %v", err) + } + run, err := s.TriggerNow(ctx, jobID) + if err != nil { + t.Fatalf("trigger now: %v", err) + } + + limits := RunLimits{MaxBytes: 10} + runErr := RunWithLimits(ctx, limits, func(ctx context.Context, bytes *ByteCounter) error { + return bytes.Add(20) + }) + if !IsPermanent(runErr) { + t.Fatalf("runErr = %v, erwartet Permanent", runErr) + } + + policy := RetryPolicy{MaxAttempts: 10, BaseDelay: time.Millisecond, MaxDelay: time.Millisecond} + status, err := s.CompleteRunWithRetry(ctx, run.ID, runErr, policy, time.Now()) + if err != nil { + t.Fatalf("complete run with retry: %v", err) + } + if status != RunStatusFailed { + t.Fatalf("status = %q, erwartet failed (terminal trotz 10 verbleibender Versuche)", status) + } + + due, err := s.ClaimDueRetries(ctx, time.Now().Add(time.Hour)) + if err != nil { + t.Fatalf("claim due retries: %v", err) + } + if len(due) != 0 { + t.Fatalf("claimed %d retries nach Ressourcenlimit-Abbruch, erwartet 0", len(due)) + } +} diff --git a/connect/internal/thirdpartyrate/thirdpartyrate.go b/connect/internal/thirdpartyrate/thirdpartyrate.go new file mode 100644 index 00000000..18565682 --- /dev/null +++ b/connect/internal/thirdpartyrate/thirdpartyrate.go @@ -0,0 +1,131 @@ +// Package thirdpartyrate rate-limits outgoing requests to a single +// third-party system (ENG-08 AC3): a token bucket keyed by connector +// instance, so a connector stuck in a request/retry loop against its +// third system waits for its own slots instead of ever touching another +// connector's or tenant's capacity. +package thirdpartyrate + +import ( + "context" + "fmt" + "net/http" + "sync" + "time" +) + +// Limits configures one key's token bucket: refilled at RequestsPerSecond, +// capped at Burst. +type Limits struct { + RequestsPerSecond float64 + Burst int +} + +type bucket struct { + mu sync.Mutex + limits Limits + tokens float64 + lastFill time.Time +} + +// Limiter holds one independent token bucket per key. Zero value is not +// usable; construct with New. +type Limiter struct { + mu sync.Mutex + def Limits + buckets map[string]*bucket +} + +// New creates a Limiter whose keys use def unless overridden via SetLimit. +func New(def Limits) *Limiter { + return &Limiter{def: def, buckets: make(map[string]*bucket)} +} + +// SetLimit overrides the bucket configuration for one key (e.g. one +// connector instance) without affecting any other key. +func (l *Limiter) SetLimit(key string, limits Limits) { + b := l.bucketFor(key) + b.mu.Lock() + b.limits = limits + b.mu.Unlock() +} + +func (l *Limiter) bucketFor(key string) *bucket { + l.mu.Lock() + defer l.mu.Unlock() + b, ok := l.buckets[key] + if !ok { + b = &bucket{limits: l.def, tokens: float64(l.def.Burst), lastFill: time.Now()} + l.buckets[key] = b + } + return b +} + +// Wait blocks until a request slot for key is available, or ctx is done. +// Independent keys never block each other (Prüfung 3): each has its own +// bucket and mutex. +func (l *Limiter) Wait(ctx context.Context, key string) error { + b := l.bucketFor(key) + for { + wait, ok := b.take() + if ok { + return nil + } + timer := time.NewTimer(wait) + select { + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + case <-timer.C: + } + } +} + +// take refills the bucket for elapsed time and, if a token is available, +// consumes one and returns (0, true); otherwise it returns how long the +// caller should wait before trying again. +func (b *bucket) take() (time.Duration, bool) { + b.mu.Lock() + defer b.mu.Unlock() + + rate := b.limits.RequestsPerSecond + if rate <= 0 { + rate = 1 + } + burst := float64(b.limits.Burst) + if burst <= 0 { + burst = 1 + } + + now := time.Now() + b.tokens += now.Sub(b.lastFill).Seconds() * rate + if b.tokens > burst { + b.tokens = burst + } + b.lastFill = now + + if b.tokens >= 1 { + b.tokens-- + return 0, true + } + return time.Duration((1 - b.tokens) / rate * float64(time.Second)), false +} + +// RoundTripper makes every outgoing request through Next wait for a slot on +// Limiter under Key before being sent, without the connector's own +// request-building code needing to know about rate limiting. +type RoundTripper struct { + Next http.RoundTripper + Limiter *Limiter + Key string +} + +func (rt *RoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + if err := rt.Limiter.Wait(req.Context(), rt.Key); err != nil { + return nil, fmt.Errorf("thirdpartyrate: warten auf slot fuer %q: %w", rt.Key, err) + } + next := rt.Next + if next == nil { + next = http.DefaultTransport + } + return next.RoundTrip(req) +} diff --git a/connect/internal/thirdpartyrate/thirdpartyrate_test.go b/connect/internal/thirdpartyrate/thirdpartyrate_test.go new file mode 100644 index 00000000..11669b35 --- /dev/null +++ b/connect/internal/thirdpartyrate/thirdpartyrate_test.go @@ -0,0 +1,100 @@ +package thirdpartyrate + +import ( + "context" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +// AC3 / Pruefung 3 (Kern): Ausgehende Anfragen an ein Drittsystem sind +// ratenbegrenzt. +func TestWait_LimitsRequestsPerSecond(t *testing.T) { + l := New(Limits{RequestsPerSecond: 10, Burst: 1}) + ctx := context.Background() + + if err := l.Wait(ctx, "connector-a"); err != nil { + t.Fatalf("erste Anfrage: %v", err) + } + start := time.Now() + if err := l.Wait(ctx, "connector-a"); err != nil { + t.Fatalf("zweite Anfrage: %v", err) + } + elapsed := time.Since(start) + if elapsed < 80*time.Millisecond { + t.Fatalf("zweite Anfrage kam nach %s, erwartet ~100ms Wartezeit (10 req/s)", elapsed) + } +} + +// AC3 / Pruefung 3 (Lasttest): ein Connector mit Drittsystem-Fehlverhalten +// (dauerhafte Anfragen ohne Limit-Respekt) blockiert nicht die Anfragen +// eines anderen Connectors — jeder Schluessel hat seinen eigenen Bucket. +func TestWait_MisbehavingConnectorDoesNotBlockOtherConnector(t *testing.T) { + l := New(Limits{RequestsPerSecond: 1, Burst: 1}) // sehr eng fuer den "boesen" Connector + l.SetLimit("connector-good", Limits{RequestsPerSecond: 1000, Burst: 1000}) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + // "boesartiger" Connector haemmert unbegrenzt gegen sein enges Limit. + badCtx, badCancel := context.WithCancel(context.Background()) + defer badCancel() + go func() { + for { + if err := l.Wait(badCtx, "connector-bad"); err != nil { + return + } + } + }() + + var done int32 + start := time.Now() + for i := 0; i < 500; i++ { + if err := l.Wait(ctx, "connector-good"); err != nil { + t.Fatalf("guter Connector blockiert (nach %d von 500): %v", i, err) + } + } + atomic.StoreInt32(&done, 1) + if elapsed := time.Since(start); elapsed > time.Second { + t.Fatalf("500 Anfragen des unabhaengigen Connectors dauerten %s, erwartet << 1s", elapsed) + } +} + +func TestRoundTripper_WaitsBeforeForwarding(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + l := New(Limits{RequestsPerSecond: 5, Burst: 1}) + client := &http.Client{Transport: &RoundTripper{Limiter: l, Key: "instance-1"}} + + resp, err := client.Get(srv.URL) + if err != nil { + t.Fatalf("erste Anfrage: %v", err) + } + _ = resp.Body.Close() + + start := time.Now() + resp, err = client.Get(srv.URL) + if err != nil { + t.Fatalf("zweite Anfrage: %v", err) + } + _ = resp.Body.Close() + if elapsed := time.Since(start); elapsed < 100*time.Millisecond { + t.Fatalf("zweite Anfrage kam nach %s, erwartet Wartezeit durch Rate-Limit", elapsed) + } +} + +func TestWait_ContextCancelled_ReturnsError(t *testing.T) { + l := New(Limits{RequestsPerSecond: 1, Burst: 1}) + _ = l.Wait(context.Background(), "k") // ersten Slot verbrauchen + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + if err := l.Wait(ctx, "k"); err == nil { + t.Fatal("erwarteter Kontext-Fehler ausgeblieben") + } +} -- 2.47.3