feat(mail): ARC-09 Postfach-Quota (unabhängig von Core LIC-05)

mailboxconfig (IMP-07) bekommt eine quota_bytes-Spalte statt einer
eigenen Tabelle — ein Postfach ist bereits eindeutig über
(tenant_slug, name) identifiziert. SetQuotaBytes/LimitBytes, 0 =
unbegrenzt (Standardwert, keine Migration bestehender Postfächer
nötig). LimitBytes erfüllt strukturell quota.LimitProvider.

storage.ArchiveMailboxPrefix (ARC-04-Ergänzung, Präfix ALLER Jahre
eines Postfachs) und storage.UsageCounter: realer Speicherverbrauch
durch echtes S3-Listing im physisch getrennten Mandanten-Bucket
(ARC-06) — kein separat gepflegter Zählerstand.

Neues Paket mail/internal/quota: Checker verbindet LimitProvider und
UsageProvider. Kein konfiguriertes Limit = immer erlaubt (Core-LIC-05-
Quota läuft unabhängig weiter — beide Ebenen bewusst unabhängig
durchgesetzt, bekannter Fehler vermieden).

smtp.QuotaChecker (schmale Schnittstelle, keine Paketkopplung an
quota) wird in handleRcptTo geprüft, VOR der Datenübertragung: 552
(RFC 5321 "exceeded storage allocation") bei Überschreitung, Session
bleibt nutzbar. nil-Checker erhält bisheriges Verhalten unverändert.

Alle drei Pflichtprüfungen mit echten Nachweisen: Quota-Überschreitung
liefert 552, Session bleibt funktionsfähig; ein anderes Postfach
desselben Tenants läuft währenddessen vollständig normal durch;
vollständiger Ende-zu-Ende-Integrationstest gegen reale Postgres- und
MinIO-Instanzen — 5000 echte Bytes abgelegt, real gemessen, Limit
knapp darunter/darüber gesetzt, SMTP reagiert jeweils korrekt auf den
tatsächlichen gemessenen Wert. Dabei einen echten Cleanup-Fehler
gefunden und behoben (defer schloss den Pool vor dem zugehörigen
t.Cleanup, verwaiste Testdaten blieben zurück).

go build/go vet/golangci-lint clean, gesamtes Mail-Modul
regressionsfrei getestet.

mail/docs/ARC-09-PRUEFPROTOKOLL.md

@@ -0,0 +1,112 @@
+# ARC-09 — Postfach-Quota: Prüfprotokoll
+
+Datum: 2026-09-02
+Host: 192.168.1.131 (Build/Test/Lint), rsync + ssh
+Pakete: `mail/internal/quota` (neu), `mail/internal/storage` (`usagecounter.go`, `archivekey.go` erweitert), `mail/internal/mailboxconfig` (erweitert), `mail/internal/smtp` (erweitert)
+
+## Umsetzung
+
+**Konfiguriertes Limit** (Akzeptanzkriterium 1): `mailboxconfig`
+(IMP-07) bekommt eine neue Spalte `quota_bytes` (Migration
+`0002_mail_mailboxes_quota.sql`, idempotent nachgezogen) statt einer
+eigenen Tabelle — ein Postfach ist bereits eindeutig über
+`(tenant_slug, name)` identifiziert. `Store.SetQuotaBytes`/`LimitBytes`
+(`0` = unbegrenzt, Standardwert, keine Migration bestehender
+Postfächer nötig). `LimitBytes` erfüllt strukturell `quota.
+LimitProvider` — eigenständig von der tenant-weiten Core-LIC-05-Quota.
+
+**Realer Verbrauch** (Pflichtprüfung 3): `storage.UsageCounter`
+summiert die TATSÄCHLICHE Objektgröße aller Objekte unter
+`storage.ArchiveMailboxPrefix(mailbox)` (neu, ARC-04-Ergänzung — Präfix
+ALLER Jahre eines Postfachs) im physisch getrennten Mandanten-Bucket
+(ARC-06) — kein separat gepflegter Zählerstand, der von der
+tatsächlichen Ablage abweichen könnte. Erfüllt strukturell `quota.
+UsageProvider`.
+
+**Verknüpfung**: `quota.Checker` (neues Paket) verbindet
+`LimitProvider` und `UsageProvider`: kein konfiguriertes Limit =
+immer erlaubt (Core-LIC-05-Quota läuft unabhängig weiter, bekannter
+Fehler bewusst vermieden — beide Ebenen unabhängig durchgesetzt).
+
+**SMTP-Durchsetzung** (Akzeptanzkriterium 2): `smtp.QuotaChecker`
+(schmale Schnittstelle, keine Paketkopplung an `quota`) wird in
+`handleRcptTo` geprüft — VOR der Datenübertragung, nicht erst nach
+vollständigem DATA-Empfang. Bei Überschreitung: `552` (RFC 5321
+"exceeded storage allocation"), Session bleibt nutzbar. Der Empfänger
+(RCPT-TO-Adresse) ist der Postfachbezug — dasselbe `mailbox`-Feld wie
+`storage.ArchiveKey`/`mailboxconfig`. `quotaChecker == nil` erhält das
+bisherige Verhalten unverändert (Rückwärtskompatibilität zu
+ING-01..QA-04).
+
+## Pflichtprüfung 1: Postfach-Quota erreicht, neue eingehende Mail wird mit korrekter SMTP-Fehlermeldung abgelehnt
+
+`TestRcptTo_QuotaExceededRejectedWithCorrectSMTPError`: RCPT TO an ein
+als "am Limit" markiertes Postfach liefert `552` mit erkennbarer
+Quota-Fehlermeldung; Session bleibt danach funktionsfähig (`NOOP` →
+`250`); der Sink bekommt keine Nachricht.
+
+Ergebnis: **BESTANDEN**.
+
+## Pflichtprüfung 2: anderes Postfach desselben Tenants empfängt weiterhin normal, während eines am Limit ist
+
+`TestRcptTo_OtherMailboxUnaffectedWhenOneAtLimit`: zwei unabhängige
+SMTP-Transaktionen desselben Tenants — die erste (Postfach am Limit)
+wird mit `552` abgelehnt, die zweite (anderes Postfach, kein Limit)
+läuft vollständig durch (`250`/`354`/`250`), die Nachricht kommt real
+beim Sink an.
+
+Ergebnis: **BESTANDEN**.
+
+## Pflichtprüfung 3: Verbrauchsanzeige je Postfach im Test korrekt gegen tatsächliche Größe geprüft
+
+`TestIntegration_UsageDisplayMatchesRealSizeAndEnforcesQuota`
+(vollständiger Ende-zu-Ende-Integrationstest, echte Postgres- und
+MinIO-Instanz): 5000 Bytes real in den ARC-06-Bucket eines real
+provisionierten Mandanten geschrieben, `storage.UsageCounter.
+UsageBytes` gemessen — der gemessene Wert liegt bei/über der
+tatsächlich geschriebenen Größe (das Prüfsummen-Sidecar-Objekt aus
+ARC-01 zählt strukturell mit, daher `>=` statt `==` geprüft). Limit
+knapp UNTER dem real gemessenen Verbrauch gesetzt → `RCPT TO` liefert
+real `552`; Limit anschließend großzügig ÜBER den Verbrauch erhöht →
+dieselbe Adresse liefert danach real `250` — die Quota-Durchsetzung
+reagiert korrekt auf den ECHTEN, gemessenen Wert, nicht auf einen
+angenommenen.
+
+Ergebnis: **BESTANDEN** (inklusive eines während der Testentwicklung
+gefundenen und behobenen Cleanup-Fehlers: `defer pool.Close()` schloss
+die Postgres-Verbindung VOR den zugehörigen `t.Cleanup`-Löschungen,
+wodurch verwaiste Registry-/Postfach-Zeilen zurückblieben — behoben
+durch `t.Cleanup(pool.Close)` statt `defer`, LIFO-Reihenfolge stellt
+sicher, dass Löschungen vor dem Verbindungsschluss laufen; durch zwei
+aufeinanderfolgende reale Testläufe bestätigt).
+
+## Akzeptanzkriterien
+
+1. **Speicherlimit ist je Postfach konfigurierbar, unabhängig von der
+   Tenant-weiten Quota aus Core LIC-05**: `mailboxconfig.
+   SetQuotaBytes`/`LimitBytes`, durch Pflichtprüfung 3 belegt.
+2. **Postfach am Limit lehnt neue eingehende Mail mit klarer,
+   protokollgerechter SMTP-Fehlermeldung ab**: `552` bei `RCPT TO`,
+   durch Pflichtprüfung 1 belegt.
+3. **Ein Postfach am Limit beeinträchtigt keine anderen Postfächer
+   desselben Tenants**: durch Pflichtprüfung 2 belegt.
+
+## Build/Vet/Lint/Test — Gesamtmodul
+
+```
+go build ./...    → OK
+go vet ./...      → OK
+golangci-lint run ./... → 0 issues
+go test ./... -p 1 (TEST_TENANT_DSN, TEST_MANTICORE_URL, TEST_S3_ENDPOINT/TEST_S3_ACCESS_KEY/TEST_S3_SECRET_KEY gesetzt) → alle Pakete ok, inkl. neuem internal/quota
+```
+
+Keine Regression — insbesondere bestehende `mailboxconfig`-Tests
+(IMP-07) bleiben nach der neuen `quota_bytes`-Spalte unverändert grün.
+
+## Ergebnis
+
+ARC-09 erfüllt alle Akzeptanzkriterien mit echten, ausgeführten
+Nachweisen — inklusive eines vollständigen Ende-zu-Ende-Integrations-
+tests gegen reale Postgres- und MinIO-Instanzen. Freigeschaltet: QA-05
+(zusammen mit ARC-07/10/INT-08, ARC-05 weiterhin extern blockiert
+durch RET-03).

mail/internal/mailboxconfig/migrations/0002_mail_mailboxes_quota.sql

@@ -0,0 +1 @@
+ALTER TABLE mail_mailboxes ADD COLUMN IF NOT EXISTS quota_bytes BIGINT NOT NULL DEFAULT 0;

mail/internal/mailboxconfig/store.go

@@ -28,6 +28,9 @@ import (
 //go:embed migrations/0001_mail_mailboxes.sql
 var schemaMigration string
 
+//go:embed migrations/0002_mail_mailboxes_quota.sql
+var quotaMigration string
+
 // ErrNotFound wird geliefert, wenn kein Postfach mit den angegebenen
 // Bezugsdaten existiert.
 var ErrNotFound = errors.New("mailboxconfig: postfach nicht gefunden")
@@ -59,11 +62,15 @@ func NewStore(pool *pgxpool.Pool, cryptoSvc *crypto.Service) *Store {
 	return &Store{pool: pool, crypto: cryptoSvc}
 }
 
-// EnsureSchema legt die Tabelle an, falls sie noch nicht existiert.
+// EnsureSchema legt die Tabelle an, falls sie noch nicht existiert, und
+// zieht die quota_bytes-Spalte nach (ARC-09).
 func (s *Store) EnsureSchema(ctx context.Context) error {
 	if _, err := s.pool.Exec(ctx, schemaMigration); err != nil {
 		return fmt.Errorf("mailboxconfig: schema anlegen: %w", err)
 	}
+	if _, err := s.pool.Exec(ctx, quotaMigration); err != nil {
+		return fmt.Errorf("mailboxconfig: quota-spalte anlegen: %w", err)
+	}
 	return nil
 }
 
@@ -183,6 +190,46 @@ func (s *Store) Delete(ctx context.Context, tenantSlug string, id int64) error {
 	return nil
 }
 
+// SetQuotaBytes setzt das Speicherlimit EINES Postfachs, identifiziert
+// über seinen (innerhalb des Mandanten eindeutigen) Namen (ARC-09
+// Akzeptanzkriterium 1: je Postfach konfigurierbar, unabhängig von der
+// Tenant-weiten Core-LIC-05-Quota). 0 bedeutet unbegrenzt — der
+// Standardwert für bestehende Postfächer, keine Migration nötig.
+func (s *Store) SetQuotaBytes(ctx context.Context, tenantSlug, name string, quotaBytes int64) error {
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE mail_mailboxes SET quota_bytes = $3, updated_at = now()
+		WHERE tenant_slug = $1 AND name = $2
+	`, tenantSlug, name, quotaBytes)
+	if err != nil {
+		return fmt.Errorf("mailboxconfig: quota setzen: %w", err)
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrNotFound
+	}
+	return nil
+}
+
+// LimitBytes erfüllt quota.LimitProvider (ARC-09): configured=false
+// bedeutet "kein Limit gesetzt, unbegrenzt" — sowohl wenn kein
+// Postfach dieses Namens für tenantSlug existiert, als auch wenn
+// quota_bytes auf 0 (den Standardwert) steht.
+func (s *Store) LimitBytes(ctx context.Context, tenantSlug, name string) (limitBytes int64, configured bool, err error) {
+	var quotaBytes int64
+	err = s.pool.QueryRow(ctx, `
+		SELECT quota_bytes FROM mail_mailboxes WHERE tenant_slug = $1 AND name = $2
+	`, tenantSlug, name).Scan(&quotaBytes)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return 0, false, nil
+		}
+		return 0, false, fmt.Errorf("mailboxconfig: quota lesen: %w", err)
+	}
+	if quotaBytes <= 0 {
+		return 0, false, nil
+	}
+	return quotaBytes, true, nil
+}
+
 // GetDecryptedPassword entschlüsselt das Postfach-Passwort — separater,
 // bewusster Aufruf statt Bestandteil von List/Get, damit Zugangsdaten
 // nicht beiläufig mitgeliefert werden.

mail/internal/quota/quota.go

@@ -0,0 +1,69 @@
+// Package quota implementiert ARC-09: konfigurierbares Speicherlimit
+// je einzelnem Postfach — eine ANDERE Ebene als die tenant-weite Quota
+// aus Core LIC-05 (bekannter Fehler zu vermeiden: beide Ebenen müssen
+// unabhängig durchgesetzt werden, ein Tenant kann insgesamt genug
+// Kontingent haben, während ein einzelnes Postfach überläuft).
+package quota
+
+import "context"
+
+// LimitProvider liefert das konfigurierte Speicherlimit eines
+// Postfachs (Akzeptanzkriterium 1). configured=false bedeutet: kein
+// Limit gesetzt, also unbegrenzt — erfüllt von
+// mail/internal/mailboxconfig.Store.LimitBytes.
+type LimitProvider interface {
+	LimitBytes(ctx context.Context, tenantSlug, mailbox string) (limitBytes int64, configured bool, err error)
+}
+
+// UsageProvider liefert den TATSÄCHLICHEN, real gemessenen
+// Speicherverbrauch eines Postfachs — erfüllt von
+// mail/internal/storage.UsageCounter.
+type UsageProvider interface {
+	UsageBytes(ctx context.Context, tenantSlug, mailbox string) (usedBytes int64, err error)
+}
+
+// Result ist das Ergebnis einer Quota-Prüfung.
+type Result struct {
+	Allowed    bool
+	UsedBytes  int64
+	LimitBytes int64
+}
+
+// Checker verknüpft konfiguriertes Limit und realen Verbrauch.
+type Checker struct {
+	limits LimitProvider
+	usage  UsageProvider
+}
+
+func NewChecker(limits LimitProvider, usage UsageProvider) *Checker {
+	return &Checker{limits: limits, usage: usage}
+}
+
+// Check prüft, ob mailbox innerhalb seines konfigurierten Limits ist.
+// Kein konfiguriertes Limit bedeutet immer "erlaubt" — die
+// Tenant-weite Quota (Core LIC-05) läuft unabhängig davon weiter.
+func (c *Checker) Check(ctx context.Context, tenantSlug, mailbox string) (Result, error) {
+	limit, configured, err := c.limits.LimitBytes(ctx, tenantSlug, mailbox)
+	if err != nil {
+		return Result{}, err
+	}
+	if !configured {
+		return Result{Allowed: true}, nil
+	}
+	used, err := c.usage.UsageBytes(ctx, tenantSlug, mailbox)
+	if err != nil {
+		return Result{}, err
+	}
+	return Result{Allowed: used < limit, UsedBytes: used, LimitBytes: limit}, nil
+}
+
+// Allowed ist die schmale Form von Check für Aufrufer, die nur die
+// Ja/Nein-Entscheidung brauchen (z. B. mail/internal/smtp.QuotaChecker,
+// Akzeptanzkriterium 2).
+func (c *Checker) Allowed(ctx context.Context, tenantSlug, mailbox string) (bool, error) {
+	result, err := c.Check(ctx, tenantSlug, mailbox)
+	if err != nil {
+		return false, err
+	}
+	return result.Allowed, nil
+}

mail/internal/quota/quota_test.go

@@ -0,0 +1,56 @@
+package quota
+
+import (
+	"context"
+	"testing"
+)
+
+type fakeLimits struct {
+	limitBytes int64
+	configured bool
+}
+
+func (f fakeLimits) LimitBytes(context.Context, string, string) (int64, bool, error) {
+	return f.limitBytes, f.configured, nil
+}
+
+type fakeUsage struct {
+	usedBytes int64
+}
+
+func (f fakeUsage) UsageBytes(context.Context, string, string) (int64, error) {
+	return f.usedBytes, nil
+}
+
+func TestCheck_UnconfiguredLimitAlwaysAllowed(t *testing.T) {
+	c := NewChecker(fakeLimits{configured: false}, fakeUsage{usedBytes: 1_000_000_000})
+	result, err := c.Check(context.Background(), "mandant-a", "postfach-x")
+	if err != nil {
+		t.Fatalf("check: %v", err)
+	}
+	if !result.Allowed {
+		t.Fatalf("erwartete erlaubt ohne konfiguriertes limit, habe: %+v", result)
+	}
+}
+
+func TestCheck_UsageAtOrAboveLimitRejected(t *testing.T) {
+	c := NewChecker(fakeLimits{limitBytes: 1000, configured: true}, fakeUsage{usedBytes: 1000})
+	result, err := c.Check(context.Background(), "mandant-a", "postfach-x")
+	if err != nil {
+		t.Fatalf("check: %v", err)
+	}
+	if result.Allowed {
+		t.Fatalf("erwartete ablehnung bei verbrauch == limit, habe: %+v", result)
+	}
+}
+
+func TestCheck_UsageBelowLimitAllowed(t *testing.T) {
+	c := NewChecker(fakeLimits{limitBytes: 1000, configured: true}, fakeUsage{usedBytes: 999})
+	result, err := c.Check(context.Background(), "mandant-a", "postfach-x")
+	if err != nil {
+		t.Fatalf("check: %v", err)
+	}
+	if !result.Allowed {
+		t.Fatalf("erwartete erlaubt bei verbrauch unter limit, habe: %+v", result)
+	}
+}

mail/internal/smtp/arc09_integration_test.go

@@ -0,0 +1,215 @@
+// Integrationstest (ARC-09): echte Postgres-Instanz (mailboxconfig.Store)
+// und echte MinIO-Instanz (storage.UsageCounter) — dieselbe
+// Umgebungsvariablen-Konvention wie mail/internal/storage
+// (TEST_S3_...) und mail/internal/mailboxconfig (TEST_TENANT_DSN).
+package smtp
+
+import (
+	"context"
+	"net"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/crypto"
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/mailboxconfig"
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/protoguard"
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/quota"
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/storage"
+)
+
+type fixedKEKProvider struct{}
+
+func (fixedKEKProvider) TenantKEK(context.Context, string) ([]byte, error) {
+	return make([]byte, crypto.KEKSize), nil
+}
+
+// TestIntegration_UsageDisplayMatchesRealSizeAndEnforcesQuota ist die
+// geforderte Pflichtprüfung 3 (ARC-09): Verbrauchsanzeige je Postfach
+// im Test korrekt gegen tatsächliche Größe geprüft — UND end-to-end:
+// echte Bytes real in einen echten, ARC-06-provisionierten Bucket
+// geschrieben, echte Verbrauchsmessung (S3-Listing), echtes
+// konfiguriertes Limit in Postgres, echte SMTP-RCPT-TO-Ablehnung, sobald
+// der reale Verbrauch das reale Limit erreicht.
+func TestIntegration_UsageDisplayMatchesRealSizeAndEnforcesQuota(t *testing.T) {
+	dsn := os.Getenv("TEST_TENANT_DSN")
+	s3Endpoint := os.Getenv("TEST_S3_ENDPOINT")
+	if dsn == "" || s3Endpoint == "" {
+		t.Skip("TEST_TENANT_DSN/TEST_S3_ENDPOINT nicht gesetzt, Integrationstest übersprungen")
+	}
+	ctx := context.Background()
+	tenant := "mandant-arc09-integration"
+	mailbox := "rechnungen@example.com"
+
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("pool: %v", err)
+	}
+	t.Cleanup(pool.Close) // siehe registryPool weiter unten: Cleanup statt defer
+	mbStore := mailboxconfig.NewStore(pool, crypto.NewService(fixedKEKProvider{}))
+	if err := mbStore.EnsureSchema(ctx); err != nil {
+		t.Fatalf("mailboxconfig schema: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM mail_mailboxes WHERE tenant_slug = $1`, tenant)
+	})
+
+	mailboxID, err := mbStore.Create(ctx, tenant, mailboxconfig.CreateInput{
+		Name: mailbox, IMAPHost: "imap.example.com", IMAPPort: 993,
+		IMAPUsername: "user", Password: "geheim", FolderSelection: []string{"INBOX"}, IntervalSeconds: 300,
+	})
+	if err != nil {
+		t.Fatalf("postfach anlegen: %v", err)
+	}
+	_ = mailboxID
+
+	s3Admin, err := storage.NewS3AdminClient(ctx, "us-east-1", s3Endpoint, os.Getenv("TEST_S3_ACCESS_KEY"), os.Getenv("TEST_S3_SECRET_KEY"), true)
+	if err != nil {
+		t.Fatalf("s3-admin-client: %v", err)
+	}
+	registryPool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("registry-pool: %v", err)
+	}
+	// t.Cleanup statt defer: muss NACH dem unten registrierten
+	// Lösch-Cleanup schließen (t.Cleanup läuft LIFO, defer würde VOR
+	// jedem t.Cleanup laufen und den Pool bereits schließen, bevor die
+	// Aufräum-Löschung ihn benutzen kann — genau der Fehler, der beim
+	// ersten Lauf dieses Tests eine verwaiste Registry-Zeile hinterließ).
+	t.Cleanup(registryPool.Close)
+	if _, err := registryPool.Exec(ctx, `
+		CREATE TABLE IF NOT EXISTS tenants (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), slug TEXT NOT NULL UNIQUE, name TEXT NOT NULL,
+			db_dsn TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'active', created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+		)
+	`); err != nil {
+		t.Fatalf("tenants-tabelle: %v", err)
+	}
+	bucket, err := storage.ProvisionTenant(ctx, registryPool, s3Admin, tenant, "ARC-09 Integrationstest", "postgresql://x")
+	if err != nil {
+		t.Fatalf("ProvisionTenant: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = registryPool.Exec(context.Background(), `DELETE FROM tenants WHERE slug = $1`, tenant)
+	})
+	t.Cleanup(func() {
+		ctx := context.Background()
+		out, err := s3Admin.ListObjectsV2(ctx, &s3.ListObjectsV2Input{Bucket: aws.String(bucket)})
+		if err == nil {
+			for _, obj := range out.Contents {
+				_, _ = s3Admin.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(bucket), Key: obj.Key})
+			}
+		}
+		_, _ = s3Admin.DeleteBucket(ctx, &s3.DeleteBucketInput{Bucket: aws.String(bucket)})
+	})
+
+	usageReporter := noopUsageReporterForTest{}
+	storageSvc := storage.NewService(mustS3Driver(t, ctx, bucket, s3Endpoint), usageReporter, tenant)
+
+	// Echte 5000 Bytes real ablegen (unter ARC-04s Postfach-Präfix).
+	const contentSize = 5000
+	content := strings.Repeat("A", contentSize)
+	key := storage.ArchiveKey(mailbox, time.Now(), "msg-arc09-1", 0)
+	if _, err := storageSvc.Put(ctx, key, strings.NewReader(content), int64(len(content)), "text/plain"); err != nil {
+		t.Fatalf("objekt ablegen: %v", err)
+	}
+	t.Cleanup(func() {
+		_ = storageSvc.Delete(context.Background(), key, int64(len(content)))
+	})
+
+	usageCounter := storage.NewUsageCounter(s3Admin)
+
+	// Echte Verbrauchsmessung gegen die tatsächliche Größe geprüft
+	// (Pflichtprüfung 3, wörtlich).
+	measured, err := usageCounter.UsageBytes(ctx, tenant, mailbox)
+	if err != nil {
+		t.Fatalf("UsageBytes: %v", err)
+	}
+	if measured < contentSize {
+		t.Fatalf("gemessener verbrauch %d liegt unter der tatsächlich abgelegten größe %d (checksum-sidecar zählt zusätzlich mit, muss also >= sein)", measured, contentSize)
+	}
+
+	// Limit knapp UNTER dem tatsächlichen Verbrauch setzen — Postfach
+	// muss jetzt als "am Limit" gelten.
+	if err := mbStore.SetQuotaBytes(ctx, tenant, mailbox, measured-1); err != nil {
+		t.Fatalf("quota setzen: %v", err)
+	}
+
+	checker := quota.NewChecker(mbStore, usageCounter)
+	srv := NewServerWithMaxMessageBytesTLSLoggerRateLimitGuardAndQuota(
+		&fakeSink{}, defaultMaxMessageBytes, nil, nil, tenant, nil, protoguard.DefaultConfig(), checker,
+	)
+	addr, stop := startRealServer(t, srv)
+	defer stop()
+
+	c := dial(t, addr)
+	defer c.close()
+	c.send(t, "EHLO client.example.com")
+	for {
+		line := c.readLine(t)
+		if strings.HasPrefix(line, "250 ") {
+			break
+		}
+	}
+	c.send(t, "MAIL FROM:<absender@example.com>")
+	resp := c.send(t, "RCPT TO:<"+mailbox+">")
+	if code(resp) != "552" {
+		t.Fatalf("erwartete 552 (reales limit real überschritten), habe: %q", resp)
+	}
+
+	// Limit großzügig ÜBER dem tatsächlichen Verbrauch setzen — dasselbe
+	// Postfach muss jetzt wieder akzeptiert werden.
+	if err := mbStore.SetQuotaBytes(ctx, tenant, mailbox, measured*10); err != nil {
+		t.Fatalf("quota erhöhen: %v", err)
+	}
+	c2 := dial(t, addr)
+	defer c2.close()
+	c2.send(t, "EHLO client.example.com")
+	for {
+		line := c2.readLine(t)
+		if strings.HasPrefix(line, "250 ") {
+			break
+		}
+	}
+	c2.send(t, "MAIL FROM:<absender@example.com>")
+	resp2 := c2.send(t, "RCPT TO:<"+mailbox+">")
+	if code(resp2) != "250" {
+		t.Fatalf("erwartete 250 nach limit-erhöhung, habe: %q", resp2)
+	}
+}
+
+type noopUsageReporterForTest struct{}
+
+func (noopUsageReporterForTest) Report(context.Context, string, string, int64) error { return nil }
+
+func mustS3Driver(t *testing.T, ctx context.Context, bucket, endpoint string) *storage.S3Driver {
+	t.Helper()
+	driver, err := storage.NewS3Driver(ctx, bucket, "us-east-1", endpoint, os.Getenv("TEST_S3_ACCESS_KEY"), os.Getenv("TEST_S3_SECRET_KEY"), true)
+	if err != nil {
+		t.Fatalf("s3-driver: %v", err)
+	}
+	return driver
+}
+
+func startRealServer(t *testing.T, srv *Server) (addr string, stop func()) {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listener: %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		_ = srv.Serve(ctx, listener)
+		close(done)
+	}()
+	return listener.Addr().String(), func() {
+		cancel()
+		<-done
+	}
+}

mail/internal/smtp/arc09_quota_test.go

@@ -0,0 +1,155 @@
+package smtp
+
+import (
+	"context"
+	"net"
+	"strings"
+	"testing"
+
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/protoguard"
+)
+
+// fakeQuotaChecker ist ein In-Memory-Fake für QuotaChecker — reine
+// Protokolltests brauchen keine echte Postgres-/S3-Instanz.
+type fakeQuotaChecker struct {
+	blocked map[string]bool // mailbox -> am limit
+}
+
+func (f fakeQuotaChecker) Allowed(_ context.Context, _, mailbox string) (bool, error) {
+	return !f.blocked[mailbox], nil
+}
+
+func startQuotaTestServer(t *testing.T, sink MessageSink, checker QuotaChecker) (addr string, stop func()) {
+	t.Helper()
+	srv := NewServerWithMaxMessageBytesTLSLoggerRateLimitGuardAndQuota(
+		sink, defaultMaxMessageBytes, nil, nil, "mandant-arc09", nil, protoguard.DefaultConfig(), checker,
+	)
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listener: %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		_ = srv.Serve(ctx, listener)
+		close(done)
+	}()
+	return listener.Addr().String(), func() {
+		cancel()
+		<-done
+	}
+}
+
+// TestRcptTo_QuotaExceededRejectedWithCorrectSMTPError ist die
+// geforderte Pflichtprüfung 1 (ARC-09): Postfach-Quota erreicht, neue
+// eingehende Mail wird mit korrekter SMTP-Fehlermeldung (552, RFC 5321
+// "exceeded storage allocation") abgelehnt, NICHT stillschweigend
+// verworfen.
+func TestRcptTo_QuotaExceededRejectedWithCorrectSMTPError(t *testing.T) {
+	sink := &fakeSink{}
+	checker := fakeQuotaChecker{blocked: map[string]bool{"voll@example.com": true}}
+	addr, stop := startQuotaTestServer(t, sink, checker)
+	defer stop()
+
+	c := dial(t, addr)
+	defer c.close()
+	c.send(t, "EHLO client.example.com")
+	for {
+		line := c.readLine(t)
+		if strings.HasPrefix(line, "250 ") {
+			break
+		}
+	}
+	c.send(t, "MAIL FROM:<a@example.com>")
+
+	resp := c.send(t, "RCPT TO:<voll@example.com>")
+	if code(resp) != "552" {
+		t.Fatalf("erwartete 552 (quota überschritten), habe: %q", resp)
+	}
+	if !strings.Contains(resp, "quota") {
+		t.Fatalf("erwartete erkennbare quota-fehlermeldung, habe: %q", resp)
+	}
+
+	// Session bleibt funktionsfähig — kein Verbindungsabbruch, keine
+	// stillschweigende Verwerfung.
+	if resp := c.send(t, "NOOP"); code(resp) != "250" {
+		t.Fatalf("session nach quota-ablehnung nicht mehr funktionsfähig: %q", resp)
+	}
+	if sink.count() != 0 {
+		t.Fatalf("erwartete keine angenommene nachricht bei quota-ablehnung, habe %d", sink.count())
+	}
+}
+
+// TestRcptTo_OtherMailboxUnaffectedWhenOneAtLimit ist die geforderte
+// Pflichtprüfung 2 (ARC-09): ein anderes Postfach desselben Tenants
+// empfängt weiterhin normal, während eines am Limit ist.
+func TestRcptTo_OtherMailboxUnaffectedWhenOneAtLimit(t *testing.T) {
+	sink := &fakeSink{}
+	checker := fakeQuotaChecker{blocked: map[string]bool{"voll@example.com": true}}
+	addr, stop := startQuotaTestServer(t, sink, checker)
+	defer stop()
+
+	// Erste, VOLLSTÄNDIGE Transaktion an das Postfach am Limit —
+	// abgelehnt.
+	c1 := dial(t, addr)
+	c1.send(t, "EHLO client.example.com")
+	for {
+		line := c1.readLine(t)
+		if strings.HasPrefix(line, "250 ") {
+			break
+		}
+	}
+	c1.send(t, "MAIL FROM:<a@example.com>")
+	if resp := c1.send(t, "RCPT TO:<voll@example.com>"); code(resp) != "552" {
+		t.Fatalf("erwartete 552 für volles postfach, habe: %q", resp)
+	}
+	c1.close()
+
+	// Zweite, unabhängige Transaktion an ein ANDERES Postfach desselben
+	// Tenants — muss vollständig normal durchlaufen.
+	c2 := dial(t, addr)
+	defer c2.close()
+	c2.send(t, "EHLO client.example.com")
+	for {
+		line := c2.readLine(t)
+		if strings.HasPrefix(line, "250 ") {
+			break
+		}
+	}
+	c2.send(t, "MAIL FROM:<a@example.com>")
+	if resp := c2.send(t, "RCPT TO:<frei@example.com>"); code(resp) != "250" {
+		t.Fatalf("erwartete 250 für unbeschränktes postfach, habe: %q", resp)
+	}
+	if resp := c2.send(t, "DATA"); code(resp) != "354" {
+		t.Fatalf("erwartete 354, habe: %q", resp)
+	}
+	if resp := c2.send(t, "Subject: test\r\n\r\nHallo\r\n."); code(resp) != "250" {
+		t.Fatalf("erwartete 250 nach abgeschlossener DATA, habe: %q", resp)
+	}
+	if sink.count() != 1 {
+		t.Fatalf("erwartete 1 angenommene nachricht für das freie postfach, habe %d", sink.count())
+	}
+}
+
+// TestRcptTo_NoQuotaCheckerAllowsAll stellt sicher, dass ohne
+// konfigurierten QuotaChecker (nil, Rückwärtskompatibilität) alle
+// Empfänger wie vor ARC-09 akzeptiert werden.
+func TestRcptTo_NoQuotaCheckerAllowsAll(t *testing.T) {
+	sink := &fakeSink{}
+	addr, stop := startTestServer(t, sink, defaultMaxMessageBytes)
+	defer stop()
+
+	c := dial(t, addr)
+	defer c.close()
+	c.send(t, "EHLO client.example.com")
+	for {
+		line := c.readLine(t)
+		if strings.HasPrefix(line, "250 ") {
+			break
+		}
+	}
+	c.send(t, "MAIL FROM:<a@example.com>")
+	if resp := c.send(t, "RCPT TO:<beliebig@example.com>"); code(resp) != "250" {
+		t.Fatalf("erwartete 250 ohne quotachecker, habe: %q", resp)
+	}
+}

mail/internal/smtp/commands.go

@@ -95,7 +95,7 @@ func (s *Session) handleMailFrom(arg string) bool {
 // handleRcptTo ist Teil des Envelope-Aufbaus (Akzeptanzkriterium 1):
 // jede Empfängeradresse wird vor der Annahme validiert; mehrere RCPT TO
 // sind erlaubt.
-func (s *Session) handleRcptTo(arg string) bool {
+func (s *Session) handleRcptTo(ctx context.Context, arg string) bool {
 	if s.state != MailFromSet && s.state != RcptToSet {
 		return s.reply(503, "send MAIL FROM first") == nil
 	}
@@ -108,6 +108,22 @@ func (s *Session) handleRcptTo(arg string) bool {
 		// SMTP-Fehlermeldung statt Absturz oder Verbindungsabbruch.
 		return s.reply(553, "invalid recipient address") == nil
 	}
+	if s.quotaChecker != nil {
+		// ARC-09 Akzeptanzkriterium 2: ein Postfach am Limit lehnt neue
+		// eingehende Mail mit klarer, protokollgerechter SMTP-Fehlermeldung
+		// ab (552 = RFC 5321 "exceeded storage allocation") — an RCPT TO
+		// geprüft, BEVOR die Nachricht überhaupt übertragen wird, statt
+		// nach vollständigem DATA-Empfang zu verwerfen. addr (der
+		// Empfänger) ist der Postfachbezug — dasselbe mailbox-Feld wie
+		// mail/internal/storage.ArchiveKey/mail/internal/mailboxconfig.
+		allowed, err := s.quotaChecker.Allowed(ctx, s.tenantID, addr)
+		if err != nil {
+			return s.reply(451, "unable to verify mailbox quota, try again later") == nil
+		}
+		if !allowed {
+			return s.reply(552, "mailbox quota exceeded for "+addr) == nil
+		}
+	}
 	s.to = append(s.to, addr)
 	s.state = RcptToSet
 	return s.reply(250, "OK") == nil

mail/internal/smtp/server.go

@@ -38,6 +38,7 @@ type Server struct {
 
 	tenantID      string
 	acceptLimiter *ratelimit.Limiter
+	quotaChecker  QuotaChecker
 }
 
 func NewServer(sink MessageSink) *Server {
@@ -80,6 +81,14 @@ func NewServerWithMaxMessageBytesTLSLoggerRateLimitAndGuardConfig(sink MessageSi
 	return &Server{sink: sink, maxMessageBytes: maxMessageBytes, tlsConfig: tlsConfig, logger: logger, tenantID: tenantID, acceptLimiter: acceptLimiter, guardCfg: guardCfg}
 }
 
+// NewServerWithMaxMessageBytesTLSLoggerRateLimitGuardAndQuota erlaubt
+// zusätzlich Postfach-Quota-Prüfung bei RCPT TO (ARC-09).
+// quotaChecker darf nil sein (Quota-Prüfung dann deaktiviert,
+// Rückwärtskompatibilität zu ING-01..QA-04).
+func NewServerWithMaxMessageBytesTLSLoggerRateLimitGuardAndQuota(sink MessageSink, maxMessageBytes int64, tlsConfig *tls.Config, logger *slog.Logger, tenantID string, acceptLimiter *ratelimit.Limiter, guardCfg protoguard.Config, quotaChecker QuotaChecker) *Server {
+	return &Server{sink: sink, maxMessageBytes: maxMessageBytes, tlsConfig: tlsConfig, logger: logger, tenantID: tenantID, acceptLimiter: acceptLimiter, guardCfg: guardCfg, quotaChecker: quotaChecker}
+}
+
 // Serve nimmt Verbindungen auf listener an, bis ctx beendet wird.
 func (srv *Server) Serve(ctx context.Context, listener net.Listener) error {
 	go func() {
@@ -99,7 +108,7 @@ func (srv *Server) Serve(ctx context.Context, listener net.Listener) error {
 			}
 			return fmt.Errorf("smtp: verbindung annehmen: %w", err)
 		}
-		session := newSession(conn, srv.sink, srv.maxMessageBytes, srv.tlsConfig, srv.logger, srv.tenantID, srv.acceptLimiter, srv.guardCfg)
+		session := newSession(conn, srv.sink, srv.maxMessageBytes, srv.tlsConfig, srv.logger, srv.tenantID, srv.acceptLimiter, srv.guardCfg, srv.quotaChecker)
 		go session.Serve(ctx)
 	}
 }

mail/internal/smtp/session.go

@@ -15,6 +15,15 @@ import (
 	"gitea.perlbach24.de/scripte/nexarch/mail/internal/ratelimit"
 )
 
+// QuotaChecker prüft, ob ein Postfach noch innerhalb seines
+// konfigurierten Speicherlimits ist (ARC-09). Erfüllt von
+// *quota.Checker — als schmale Schnittstelle definiert, damit dieses
+// Paket quota nicht importieren muss (keine Kopplung an dessen
+// LimitProvider/UsageProvider-Abhängigkeiten).
+type QuotaChecker interface {
+	Allowed(ctx context.Context, tenantSlug, mailbox string) (bool, error)
+}
+
 // maxCommandLineBytes begrenzt eine einzelne Kommando-/DATA-Zeile
 // (defensive Fehlerbehandlung bei nicht-konformen Gegenstellen statt
 // optimistischem Parsing, gleiche Konvention wie mail/internal/imap und
@@ -46,6 +55,7 @@ type Session struct {
 
 	tenantID      string
 	acceptLimiter *ratelimit.Limiter // ING-09, nil = kein Rate-Limiting
+	quotaChecker  QuotaChecker       // ARC-09, nil = keine Quota-Prüfung
 
 	guard *protoguard.Guard // QA-04-Nachtrag zu ING-07
 
@@ -54,7 +64,7 @@ type Session struct {
 	to    []string
 }
 
-func newSession(conn net.Conn, sink MessageSink, maxMessageBytes int64, tlsConfig *tls.Config, logger *slog.Logger, tenantID string, acceptLimiter *ratelimit.Limiter, guardCfg protoguard.Config) *Session {
+func newSession(conn net.Conn, sink MessageSink, maxMessageBytes int64, tlsConfig *tls.Config, logger *slog.Logger, tenantID string, acceptLimiter *ratelimit.Limiter, guardCfg protoguard.Config, quotaChecker QuotaChecker) *Session {
 	_, alreadyTLS := conn.(*tls.Conn)
 	return &Session{
 		conn:            conn,
@@ -67,6 +77,7 @@ func newSession(conn net.Conn, sink MessageSink, maxMessageBytes int64, tlsConfi
 		log:             protolog.NewSessionLogger(logger, "smtp"),
 		tenantID:        tenantID,
 		acceptLimiter:   acceptLimiter,
+		quotaChecker:    quotaChecker,
 		guard:           protoguard.New(guardCfg),
 		state:           Greeting,
 	}
@@ -153,7 +164,7 @@ func (s *Session) dispatch(ctx context.Context, verb, arg string) bool {
 	case "MAIL":
 		return s.handleMailFrom(arg)
 	case "RCPT":
-		return s.handleRcptTo(arg)
+		return s.handleRcptTo(ctx, arg)
 	case "DATA":
 		return s.handleData(ctx)
 	case "RSET":

mail/internal/storage/archivekey.go

@@ -48,6 +48,17 @@ func ArchiveYearPrefix(mailbox string, year int) string {
 	return "postfach/" + mailboxSegment + "/" + yearSegment + "/"
 }
 
+// ArchiveMailboxPrefix liefert den Präfix ALLER Jahre eines Postfachs
+// (ohne Jahressegment) — Grundlage für eine reale
+// Postfach-Verbrauchsmessung über alle Jahre hinweg (ARC-09).
+func ArchiveMailboxPrefix(mailbox string) string {
+	mailboxSegment := sanitizeSegment(mailbox)
+	if mailboxSegment == "" {
+		mailboxSegment = FallbackMailboxSegment
+	}
+	return "postfach/" + mailboxSegment + "/"
+}
+
 func yearOf(t time.Time) int {
 	if t.IsZero() {
 		return 0

mail/internal/storage/usagecounter.go

@@ -0,0 +1,55 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+)
+
+// UsageCounter misst den TATSÄCHLICHEN Speicherverbrauch eines
+// Postfachs (ARC-09) durch reales Auflisten der zugehörigen Objekte im
+// physisch getrennten Mandanten-Bucket (ARC-06) unter dem
+// ARC-04-Postfach-Präfix — kein separat gepflegter Zählerstand, der
+// von der tatsächlichen Ablage abweichen könnte.
+type UsageCounter struct {
+	admin *s3.Client
+}
+
+func NewUsageCounter(admin *s3.Client) *UsageCounter {
+	return &UsageCounter{admin: admin}
+}
+
+// UsageBytes summiert die reale Objektgröße aller unter
+// ArchiveMailboxPrefix(mailbox) abgelegten Objekte im Bucket von
+// tenantSlug (ARC-06: BucketNameForTenant). Erfüllt strukturell
+// quota.UsageProvider (ARC-09), ohne dass dieses Paket quota
+// importieren müsste.
+func (u *UsageCounter) UsageBytes(ctx context.Context, tenantSlug, mailbox string) (int64, error) {
+	bucket := BucketNameForTenant(tenantSlug)
+	prefix := ArchiveMailboxPrefix(mailbox)
+
+	var total int64
+	var continuationToken *string
+	for {
+		out, err := u.admin.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
+			Bucket:            aws.String(bucket),
+			Prefix:            aws.String(prefix),
+			ContinuationToken: continuationToken,
+		})
+		if err != nil {
+			return 0, fmt.Errorf("storage: postfach-verbrauch auflisten: %w", err)
+		}
+		for _, obj := range out.Contents {
+			if obj.Size != nil {
+				total += *obj.Size
+			}
+		}
+		if out.IsTruncated == nil || !*out.IsTruncated {
+			break
+		}
+		continuationToken = out.NextContinuationToken
+	}
+	return total, nil
+}