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
+```

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
+}

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))
+	}
+}

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)
+}

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")
+	}
+}