feat(PROJ-56): Last-Entzerrung für OCR und IMAP-Sync
OCR-Worker pausieren optional in konfigurierbarem Zeitfenster (paused_hours), Jobs bleiben pending statt verworfen zu werden. IMAP-Scheduler verteilt Sync-Starts via deterministischem Pro-Account-Jitter, um Lastspitzen bei vielen Postfächern mit gleichem Intervall zu vermeiden. Beides per Config opt-out, Default-Verhalten unverändert. Build + Smoke-Test auf 132 verifiziert.
This commit is contained in:
@@ -193,7 +193,12 @@ func main() {
|
|||||||
Workers: 2,
|
Workers: 2,
|
||||||
QueueSize: 1000,
|
QueueSize: 1000,
|
||||||
Logger: logger,
|
Logger: logger,
|
||||||
|
PausedHours: cfg.OCR.PausedHours, // PROJ-56: optional local-time pause window
|
||||||
})
|
})
|
||||||
|
if cfg.OCR.PausedHours != nil {
|
||||||
|
logger.Info("ocr worker: pause window configured",
|
||||||
|
"from_hour", cfg.OCR.PausedHours[0], "to_hour", cfg.OCR.PausedHours[1])
|
||||||
|
}
|
||||||
ocrWorker.Start(context.Background())
|
ocrWorker.Start(context.Background())
|
||||||
defer ocrWorker.Stop()
|
defer ocrWorker.Stop()
|
||||||
if !ocr.IsAvailable() {
|
if !ocr.IsAvailable() {
|
||||||
@@ -433,6 +438,7 @@ func main() {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
imapSched := imapstore.NewScheduler(imapSt, imapImp, logger)
|
imapSched := imapstore.NewScheduler(imapSt, imapImp, logger)
|
||||||
|
imapSched.SetJitterSeconds(cfg.IMAPScheduler.ResolvedJitterSeconds()) // PROJ-56: deterministic per-account sync spread
|
||||||
imapSched.SetAuditLogger(audlog) // PROJ-45: tenant-visible UIDVALIDITY-reset audit entries
|
imapSched.SetAuditLogger(audlog) // PROJ-45: tenant-visible UIDVALIDITY-reset audit entries
|
||||||
imapSched.Start()
|
imapSched.Start()
|
||||||
defer imapSched.Stop()
|
defer imapSched.Stop()
|
||||||
|
|||||||
@@ -54,6 +54,22 @@ imap_server:
|
|||||||
enabled: false
|
enabled: false
|
||||||
bind: "0.0.0.0:1143"
|
bind: "0.0.0.0:1143"
|
||||||
|
|
||||||
|
# PROJ-56: OCR-Last-Entzerrung (optional).
|
||||||
|
# paused_hours definiert ein lokales Zeitfenster [start, end), in dem der
|
||||||
|
# OCR-Worker NICHT verarbeitet (Aufträge bleiben als pending erhalten).
|
||||||
|
# Wrap-around über Mitternacht wird unterstützt, z.B. [22, 6] = Pause 22:00–06:00.
|
||||||
|
# Ohne Sektion / ohne paused_hours: altes Verhalten (immer aktiv).
|
||||||
|
# ocr:
|
||||||
|
# paused_hours: [8, 18] # OCR pausiert während der Geschäftszeiten
|
||||||
|
|
||||||
|
# PROJ-56: IMAP-Sync-Jitter (optional).
|
||||||
|
# jitter_seconds verteilt den tatsächlichen Sync-Start jedes Accounts
|
||||||
|
# deterministisch (abgeleitet aus Account-ID) über dieses Fenster, damit nicht
|
||||||
|
# alle Postfächer auf derselben Minutengrenze pollen.
|
||||||
|
# Weglassen der Sektion = Default 240s (4 Min). jitter_seconds: 0 = deaktiviert.
|
||||||
|
# imap_scheduler:
|
||||||
|
# jitter_seconds: 240
|
||||||
|
|
||||||
audit:
|
audit:
|
||||||
log_path: /var/archivmail/audit.log
|
log_path: /var/archivmail/audit.log
|
||||||
retention_days: 365
|
retention_days: 365
|
||||||
|
|||||||
@@ -39,6 +39,47 @@ type Config struct {
|
|||||||
Logging LoggingConfig `yaml:"logging"`
|
Logging LoggingConfig `yaml:"logging"`
|
||||||
IMAPServer IMAPServerConfig `yaml:"imap_server"`
|
IMAPServer IMAPServerConfig `yaml:"imap_server"`
|
||||||
Metrics MetricsConfig `yaml:"metrics"`
|
Metrics MetricsConfig `yaml:"metrics"`
|
||||||
|
// PROJ-56: load-spreading for background jobs.
|
||||||
|
OCR OCRConfig `yaml:"ocr"`
|
||||||
|
IMAPScheduler IMAPSchedulerConfig `yaml:"imap_scheduler"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// OCRConfig holds settings for the background OCR worker (PROJ-56).
|
||||||
|
type OCRConfig struct {
|
||||||
|
// PausedHours optionally defines a local-time window [start, end) during
|
||||||
|
// which the OCR worker pauses processing (e.g. [8, 18] = paused 08:00–18:00).
|
||||||
|
// Wrap-around windows are supported, e.g. [22, 6] = paused 22:00–06:00.
|
||||||
|
// nil / unset = never pause (legacy behaviour: process immediately).
|
||||||
|
PausedHours *[2]int `yaml:"paused_hours,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// IMAPSchedulerConfig holds settings for the automatic IMAP sync scheduler (PROJ-56).
|
||||||
|
type IMAPSchedulerConfig struct {
|
||||||
|
// JitterSeconds spreads the per-account sync start over a deterministic
|
||||||
|
// offset derived from the account ID, so N accounts on the same interval
|
||||||
|
// don't all poll on the same minute boundary.
|
||||||
|
// A pointer so the unset case (use default) is distinguishable from an
|
||||||
|
// explicit 0 (disable jitter).
|
||||||
|
// nil -> DefaultIMAPJitterSeconds (240s window)
|
||||||
|
// 0 -> jitter disabled (legacy behaviour)
|
||||||
|
// >0 -> jitter window in seconds
|
||||||
|
JitterSeconds *int `yaml:"jitter_seconds,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// DefaultIMAPJitterSeconds is the jitter window applied when imap_scheduler
|
||||||
|
// is omitted entirely from the config (4 minutes).
|
||||||
|
const DefaultIMAPJitterSeconds = 240
|
||||||
|
|
||||||
|
// ResolvedJitterSeconds returns the effective jitter window. nil falls back to
|
||||||
|
// the default; an explicit 0 (or negative) means jitter is disabled.
|
||||||
|
func (c IMAPSchedulerConfig) ResolvedJitterSeconds() int {
|
||||||
|
if c.JitterSeconds == nil {
|
||||||
|
return DefaultIMAPJitterSeconds
|
||||||
|
}
|
||||||
|
if *c.JitterSeconds < 0 {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
return *c.JitterSeconds
|
||||||
}
|
}
|
||||||
|
|
||||||
// IMAPServerConfig holds settings for the embedded read-only IMAP archive server.
|
// IMAPServerConfig holds settings for the embedded read-only IMAP archive server.
|
||||||
|
|||||||
+2
-1
@@ -72,7 +72,8 @@
|
|||||||
| PROJ-53 | Konfigurierbare Listenanzahl pro Seite | Deployed | [PROJ-53](PROJ-53-konfigurierbare-listenanzahl.md) | 2026-06-14 |
|
| PROJ-53 | Konfigurierbare Listenanzahl pro Seite | Deployed | [PROJ-53](PROJ-53-konfigurierbare-listenanzahl.md) | 2026-06-14 |
|
||||||
| PROJ-54 | Fix Listenansicht/Pagination für Rolle "user" (Nachbesserung PROJ-6/PROJ-21) | Deployed | [PROJ-54](PROJ-54-fix-listenansicht-total.md) | 2026-06-14 |
|
| PROJ-54 | Fix Listenansicht/Pagination für Rolle "user" (Nachbesserung PROJ-6/PROJ-21) | Deployed | [PROJ-54](PROJ-54-fix-listenansicht-total.md) | 2026-06-14 |
|
||||||
| PROJ-55 | Fix Tenant-Isolation für Rolle "auditor" + Audit-Log (Sicherheitsbug, DSGVO-relevant) | Deployed | [PROJ-55](PROJ-55-fix-auditor-tenant-isolation.md) | 2026-06-21 |
|
| PROJ-55 | Fix Tenant-Isolation für Rolle "auditor" + Audit-Log (Sicherheitsbug, DSGVO-relevant) | Deployed | [PROJ-55](PROJ-55-fix-auditor-tenant-isolation.md) | 2026-06-21 |
|
||||||
|
| PROJ-56 | Last-Entzerrung für Hintergrundjobs (OCR-Zeitfenster, IMAP-Sync-Jitter) | In Review | [PROJ-56](PROJ-56-last-entzerrung-hintergrundjobs.md) | 2026-06-22 |
|
||||||
|
|
||||||
<!-- Add features above this line -->
|
<!-- Add features above this line -->
|
||||||
|
|
||||||
## Next Available ID: PROJ-56
|
## Next Available ID: PROJ-57
|
||||||
|
|||||||
@@ -0,0 +1,98 @@
|
|||||||
|
# PROJ-56: Last-Entzerrung für Hintergrundjobs (OCR, IMAP-Sync)
|
||||||
|
|
||||||
|
## Status: In Review
|
||||||
|
**Created:** 2026-06-22
|
||||||
|
**Last Updated:** 2026-06-22
|
||||||
|
|
||||||
|
## Dependencies
|
||||||
|
- PROJ-35 (OCR & Anhang-Volltext-Indexierung)
|
||||||
|
- PROJ-8 (Automatischer IMAP-Sync)
|
||||||
|
|
||||||
|
## Hintergrund (Nutzerwunsch)
|
||||||
|
Vergleich mit Mailpiler: dort laufen Wartungsprozesse zeitlich entzerrt statt komplett parallel. Bei archivmail laufen OCR-Worker (sofort bei Import, 2 parallel, je tesseract-Mehrthread) und IMAP-Sync (alle Postfächer pollen unabhängig, ohne Jitter, exakt zum Intervall-Ablauf) unkoordiniert nebeneinander, was bei begrenztem RAM (z.B. 4GB-LXC) zu Lastspitzen führt.
|
||||||
|
|
||||||
|
Manticore-Reindex wurde geprüft, ist aber bereits ein rein manuelles CLI-Kommando (kein Scheduler) — keine automatische Last-Entzerrung nötig/möglich, daher außerhalb des Scopes dieses Tickets.
|
||||||
|
|
||||||
|
## Entscheidung (Nutzer, 2026-06-22)
|
||||||
|
Alle drei vorgeschlagenen Maßnahmen gewünscht, soweit technisch sinnvoll:
|
||||||
|
1. OCR zeitlich entzerren — konfigurierbares Zeitfenster, in dem OCR-Verarbeitung pausiert (z.B. Geschäftszeiten ausnehmen).
|
||||||
|
2. IMAP-Sync-Intervalle entzerren — deterministischer Jitter pro Postfach, damit nicht alle Accounts auf derselben Minutengrenze pollen.
|
||||||
|
3. Reindex in Nachtfenster — entfällt, da bereits manuell/CLI-only ohne automatischen Trigger.
|
||||||
|
|
||||||
|
## Acceptance Criteria
|
||||||
|
- [x] `config.yml` erlaubt ein optionales OCR-Zeitfenster (z.B. `ocr.paused_hours: [8, 18]` = pausiert 08:00–18:00 Uhr); ohne Konfiguration: Verhalten unverändert (sofortige Verarbeitung, wie bisher).
|
||||||
|
- [x] Außerhalb des erlaubten Fensters werden neue OCR-Aufträge weiterhin in die Queue/DB als `pending` aufgenommen, aber nicht abgearbeitet, bis das Fenster sich öffnet (kein Datenverlust, kein Crash bei vollem Queue).
|
||||||
|
- [x] IMAP-Scheduler verteilt den tatsächlichen Sync-Start jedes Accounts über einen deterministischen Jitter (abgeleitet aus Account-ID, kein Zufall bei jedem Tick) innerhalb eines konfigurierbaren Fensters (Default z.B. 0–4 Minuten), sodass bei N Accounts mit gleichem Intervall nicht alle zur gleichen Minute synchronisieren.
|
||||||
|
- [x] Bestehende Funktionalität (Sync-Intervall pro Account, manuelles "Jetzt synchronisieren") bleibt unverändert nutzbar — Jitter gilt nur für den automatischen Scheduler-Trigger.
|
||||||
|
- [x] Kein Verhalten ändert sich für Installationen ohne neue Config-Werte (Default = bisheriges Verhalten, kein Opt-in nötig für Jitter da inhärent sinnvoll, aber abschaltbar via `jitter_seconds: 0`).
|
||||||
|
|
||||||
|
## Tech Design
|
||||||
|
Übersprungen (kleine, klar umrissene Performance-Änderung, kein architektonischer Schnitt nötig) — analog zum Vorgehen bei PROJ-55.
|
||||||
|
|
||||||
|
## Implementation Notes (2026-06-22)
|
||||||
|
|
||||||
|
### Geänderte/neue Dateien
|
||||||
|
- `config/config.go`: neue Sektionen `OCRConfig` (`paused_hours *[2]int`) und `IMAPSchedulerConfig` (`jitter_seconds *int`) in `Config`. Helper `ResolvedJitterSeconds()` + Konstante `DefaultIMAPJitterSeconds = 240`.
|
||||||
|
- `internal/ocr/worker.go`: `Options.PausedHours`, Worker-Feld `pausedHours`, `isPaused(now)`, Pause-Gate im `run`-Loop.
|
||||||
|
- `internal/imap/scheduler.go`: Feld `jitterSeconds`, `SetJitterSeconds()`, `jitterOffset()`, Anpassung der Fälligkeitsbedingung (`interval + jitterOffset(acc.ID)`).
|
||||||
|
- `cmd/archivmail/main.go`: `PausedHours` an OCR-Worker durchgereicht; `imapSched.SetJitterSeconds(cfg.IMAPScheduler.ResolvedJitterSeconds())`.
|
||||||
|
- `config/config.docker.yml.example`: auskommentierte Beispielsektionen `ocr` und `imap_scheduler`.
|
||||||
|
|
||||||
|
### OCR-Pausenmechanismus
|
||||||
|
Gewählt: **Worker konsumiert die Queue gar nicht erst, solange das Pausenfenster aktiv ist.** Vor jedem Dequeue prüft jeder Worker `isPaused(time.Now())`. Bei aktiver Pause wartet er per `select` (60s-Ticker via `pauseCheckInterval`, reagiert sofort auf `done`/`ctx.Done()`) und liest erst dann wieder aus dem Channel. Vorteile:
|
||||||
|
- Kein Datenverlust: Jobs bleiben im gepufferten Channel und v.a. als `ocr_status='pending'` in PostgreSQL.
|
||||||
|
- Kein Queue-Überlauf: Der Boot-Resume-Refill ist `QueueLen`-gesteuert und legt nichts nach, sobald der Channel voll ist; neue Submits werden wie bisher non-blocking verworfen (bleiben aber `pending` in der DB und werden beim nächsten Boot-Resume/Fensteröffnen nachgezogen).
|
||||||
|
- Kein Busy-Loop: Pollintervall 60s.
|
||||||
|
- `cmd_ocr_reprocess.go` setzt `PausedHours` bewusst nicht (nil) → manueller Admin-Befehl ignoriert das Fenster und läuft sofort.
|
||||||
|
- Wrap-around-Fenster (z.B. `[22, 6]`) werden in `isPaused` unterstützt (start>end → `h>=start || h<end`); `start==end` = No-op (nie pausieren).
|
||||||
|
|
||||||
|
### IMAP-Jitter
|
||||||
|
Deterministischer Offset `accountID % jitterSeconds` Sekunden, nur aus der Account-ID abgeleitet (stabil über alle Ticks, kein `rand()`). Fälligkeit erst bei `now.Sub(lastSync) >= interval + jitterOffset`. Default 240s wenn `imap_scheduler` fehlt (`jitter_seconds`-Pointer = nil); explizit `0` deaktiviert. `TriggerSync` (manuell) bleibt unberührt.
|
||||||
|
|
||||||
|
### Offene Risiken / Edge Cases
|
||||||
|
- Mitternachts-übergreifendes Fenster `[22,6]` ist abgedeckt; reine Stunden-Granularität (kein Minutenanteil) ist bewusst einfach gehalten.
|
||||||
|
- Server-Zeitzone: `time.Now().Hour()` nutzt lokale Serverzeit — bei UTC-Servern muss das Fenster entsprechend gesetzt werden.
|
||||||
|
- Jitter verlängert das effektive Intervall um bis zu `jitter_seconds` (max +4 Min bei Default); akzeptabel, da nur Spitzenglättung bezweckt ist.
|
||||||
|
- Lokal kein `go build` möglich (kein Toolchain) — nur statische Konsistenzprüfung erfolgt; Build-Verifikation auf 192.168.1.131.
|
||||||
|
|
||||||
|
## QA Test Results (Code-Review + Build-Verifikation, 2026-06-22)
|
||||||
|
|
||||||
|
Getestet von: QA / Red-Team. Methode: statisches Code-Review der geänderten Dateien
|
||||||
|
+ isolierte Build-Verifikation auf Test-Server 192.168.1.132 (Go 1.24.4), ohne
|
||||||
|
den produktiven Checkout / Dienst zu berühren (Tarball → /tmp/archivmail-qa56 →
|
||||||
|
`go build` → `version`-Smoke-Test + `go vet` → vollständige Bereinigung).
|
||||||
|
|
||||||
|
### Build-Verifikation (192.168.1.132)
|
||||||
|
- `CGO_ENABLED=0 go build -buildvcs=false -o /tmp/archivmail-test ./cmd/archivmail/` → **EXIT 0** (20 MB Binary).
|
||||||
|
- Smoke-Test `archivmail-test version` → OK (`archivmail 0.9.1`, Modul-Liste wird ausgegeben).
|
||||||
|
- `go vet ./config/... ./internal/ocr/... ./internal/imap/... ./cmd/archivmail/...` → **EXIT 0** (keine Befunde).
|
||||||
|
- Kein Eingriff in /opt/archivmail, kein `systemctl restart`. Temp-Artefakte auf 132 entfernt.
|
||||||
|
|
||||||
|
### Acceptance Criteria
|
||||||
|
- [PASS] AC1 — Optionales OCR-Zeitfenster: `OCRConfig.PausedHours *[2]int` (`yaml:"paused_hours,omitempty"`). `nil` → `isPaused()` gibt `false`, Verhalten unverändert.
|
||||||
|
- [PASS] AC2 — Außerhalb des Fensters bleiben Jobs `pending`: Worker konsumiert die Queue im Pausenfenster nicht (`run`-Loop: bei `isPaused` nur `select` mit 60s-Ticker, kein Dequeue). Jobs bleiben im gepufferten Channel bzw. `ocr_status='pending'` in der DB; Boot-Resume zieht sie nach. Kein Datenverlust, kein Crash bei vollem Channel (`Submit` ist non-blocking; verworfene Submits bleiben `pending` in DB).
|
||||||
|
- [PASS] AC3 — Deterministischer IMAP-Jitter: `jitterOffset(accountID) = (accountID % jitterSeconds) s`, nur aus Account-ID abgeleitet, stabil über alle Ticks (kein `rand()`). Fälligkeit erst bei `now.Sub(lastSync) >= interval + jitterOffset`.
|
||||||
|
- [PASS] AC4 — Bestehende Funktionalität unverändert: `TriggerSync` (manuell) umgeht `checkAccounts`/Jitter vollständig und startet sofort (`runSyncWithRetry` direkt). Sync-Intervall pro Account bleibt Basis.
|
||||||
|
- [PASS] AC5 — Default ohne Config: OCR `paused_hours` unset → nie Pause. `imap_scheduler` ganz weggelassen → `ResolvedJitterSeconds()` = 240s Default; `jitter_seconds: 0` → Jitter aus (Legacy-Tick exakt zum Intervall).
|
||||||
|
|
||||||
|
### Prüfpunkte
|
||||||
|
1. **Datenverlust bei Pause** — PASS. Jobs werden nicht verworfen; Channel-Buffer + `pending`-Status + Boot-Resume garantieren Nachzug.
|
||||||
|
2. **Race Conditions** — PASS (in der aktuellen Verdrahtung). `pausedHours` wird nur im Konstruktor gesetzt, danach nur gelesen → keine Concurrent-Writes. `jitterSeconds` wird in `main.go` per `SetJitterSeconds()` **vor** `imapSched.Start()` gesetzt; der Loop liest erst nach `Start()`. Kein Daten-Rennen im realen Pfad. Hinweis (LOW, kein Bug): `SetJitterSeconds`/`jitterOffset` greifen unsynchronisiert auf `s.jitterSeconds` zu — würde ein künftiger Aufrufer Jitter zur Laufzeit ändern, wäre es ein Race. Aktuell nicht der Fall. `go vet` meldet nichts; ein `-race`-Test wurde nicht ausgeführt (kein Test-Harness im Scope).
|
||||||
|
3. **Edge Cases** — PASS.
|
||||||
|
- Mitternachtsfenster `[22,6]`: `start>end` → `h>=start || h<end`, korrekt (pausiert 22–05).
|
||||||
|
- `start==end`: explizit No-op (nie pausieren).
|
||||||
|
- Fehlende Config / nil-Pointer: `isPaused` prüft `pausedHours==nil`; `ResolvedJitterSeconds` prüft `JitterSeconds==nil`. Keine nil-Derefs.
|
||||||
|
- `jitter_seconds: 0`: `jitterOffset` gibt 0 → reines Intervall-Verhalten.
|
||||||
|
- Negativ: `SetJitterSeconds` und `ResolvedJitterSeconds` klemmen `<0` auf 0.
|
||||||
|
4. **Backward-Compat** — PASS (siehe AC5). Default-Pfad identisch zum Alt-Verhalten, mit einer bewussten Ausnahme: ohne `imap_scheduler`-Sektion ist nun **240s Jitter aktiv** (laut Spec gewollt, „abschaltbar via jitter_seconds: 0"). Verlängert das effektive Sync-Intervall um bis zu 4 Min — dokumentiert in den Implementation Notes als akzeptiert.
|
||||||
|
5. **`cmd_ocr_reprocess.go` erbt kein Pausenfenster** — PASS. Der `ocr.Options`-Block dort setzt `PausedHours` nicht (= nil) → manuelle Reprocessing-Läufe ignorieren das Fenster und laufen sofort. Verifiziert (Zeilen 103–107).
|
||||||
|
6. **Manuelles "Jetzt synchronisieren" ignoriert Jitter** — PASS. `TriggerSync` ruft `runSyncWithRetry` direkt auf, ohne `dueAfter`/`jitterOffset`-Berechnung. Verifiziert.
|
||||||
|
|
||||||
|
### Findings
|
||||||
|
- **Keine Bugs (Severity Critical/High/Medium) gefunden.**
|
||||||
|
- **LOW / INFO 1 (kein Fix nötig):** `Scheduler.jitterSeconds` ist nicht mutex-geschützt. Unkritisch, da set-before-start. Nur relevant, falls künftig eine Laufzeit-Rekonfiguration eingeführt wird → dann `s.mu` o.ä. ergänzen.
|
||||||
|
- **LOW / INFO 2 (kein Fix nötig):** Pausenfenster nutzt lokale Serverzeit (`time.Now().Hour()`). Auf UTC-Servern muss das Fenster entsprechend gewählt werden — bereits in den Implementation Notes dokumentiert.
|
||||||
|
- **INFO 3 (vorbestehend, nicht Teil von PROJ-56):** `config.docker.yml.example` enthält weiterhin `xapian_path`/`backend: xapian`, obwohl der `StorageConfig` kein `xapian_path`-Feld hat und der Default-Backend Manticore ist. Unkritisch (YAML-Unmarshal ignoriert unbekannte Keys), aber irreführend. Außerhalb des Scopes dieses Tickets.
|
||||||
|
|
||||||
|
### Fazit
|
||||||
|
Production-Ready: **JA**. Alle 5 Acceptance Criteria erfüllt, alle 6 Sicherheits-/Korrektheits-Prüfpunkte bestanden, Build + vet + Smoke-Test auf 192.168.1.132 grün. Keine blockierenden Findings; nur 2 LOW-Hinweise (kein Fix nötig) und 1 vorbestehender, ticket-fremder Konfig-Hinweis.
|
||||||
@@ -26,6 +26,11 @@ type Scheduler struct {
|
|||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
running map[int64]bool // in-memory guard against concurrent syncs
|
running map[int64]bool // in-memory guard against concurrent syncs
|
||||||
|
|
||||||
|
// PROJ-56: deterministic per-account jitter window in seconds. 0 disables
|
||||||
|
// jitter. The offset for an account is derived solely from its ID so the
|
||||||
|
// effective sync time is stable across ticks (no rand() per tick).
|
||||||
|
jitterSeconds int
|
||||||
|
|
||||||
cancel context.CancelFunc
|
cancel context.CancelFunc
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -39,6 +44,32 @@ func NewScheduler(store *Store, importer *Importer, logger *slog.Logger) *Schedu
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SetJitterSeconds configures the deterministic per-account sync jitter window
|
||||||
|
// (PROJ-56). 0 disables jitter (legacy behaviour: sync exactly at interval).
|
||||||
|
// Negative values are clamped to 0.
|
||||||
|
func (s *Scheduler) SetJitterSeconds(seconds int) {
|
||||||
|
if seconds < 0 {
|
||||||
|
seconds = 0
|
||||||
|
}
|
||||||
|
s.jitterSeconds = seconds
|
||||||
|
}
|
||||||
|
|
||||||
|
// jitterOffset returns the deterministic delay added on top of the sync
|
||||||
|
// interval for a given account. The offset is in [0, jitterSeconds) and
|
||||||
|
// depends only on the account ID, so it never changes between ticks.
|
||||||
|
func (s *Scheduler) jitterOffset(accountID int64) time.Duration {
|
||||||
|
if s.jitterSeconds <= 0 {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
// Account IDs are positive sequential integers; a simple modulo spreads
|
||||||
|
// them evenly across the window. Use the absolute value defensively.
|
||||||
|
id := accountID
|
||||||
|
if id < 0 {
|
||||||
|
id = -id
|
||||||
|
}
|
||||||
|
return time.Duration(id%int64(s.jitterSeconds)) * time.Second
|
||||||
|
}
|
||||||
|
|
||||||
// SetAuditLogger wires an audit.Logger into the scheduler so that
|
// SetAuditLogger wires an audit.Logger into the scheduler so that
|
||||||
// UIDVALIDITY-reset events (PROJ-45) are persisted as tenant-visible
|
// UIDVALIDITY-reset events (PROJ-45) are persisted as tenant-visible
|
||||||
// audit entries. Optional — when nil, only structured logs are emitted.
|
// audit entries. Optional — when nil, only structured logs are emitted.
|
||||||
@@ -123,7 +154,12 @@ func (s *Scheduler) checkAccounts(ctx context.Context) {
|
|||||||
lastSync = *acc.LastSyncAt
|
lastSync = *acc.LastSyncAt
|
||||||
}
|
}
|
||||||
|
|
||||||
if now.Sub(lastSync) >= interval {
|
// PROJ-56: spread the actual sync start with a deterministic per-account
|
||||||
|
// offset so accounts sharing an interval don't all poll on the same
|
||||||
|
// minute boundary. Offset is 0 when jitter is disabled.
|
||||||
|
dueAfter := interval + s.jitterOffset(acc.ID)
|
||||||
|
|
||||||
|
if now.Sub(lastSync) >= dueAfter {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
s.running[acc.ID] = true
|
s.running[acc.ID] = true
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ import (
|
|||||||
"log/slog"
|
"log/slog"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
"archivmail/internal/index"
|
"archivmail/internal/index"
|
||||||
"archivmail/internal/storage"
|
"archivmail/internal/storage"
|
||||||
@@ -32,14 +33,28 @@ type Worker struct {
|
|||||||
wg sync.WaitGroup
|
wg sync.WaitGroup
|
||||||
workers int
|
workers int
|
||||||
langs []string
|
langs []string
|
||||||
|
|
||||||
|
// PROJ-56: optional local-time pause window [start, end). When the current
|
||||||
|
// hour falls inside it, workers stop consuming the queue (jobs stay buffered
|
||||||
|
// in the channel / as ocr_status='pending' in the DB) until it reopens.
|
||||||
|
// nil = never pause (legacy behaviour).
|
||||||
|
pausedHours *[2]int
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// pauseCheckInterval is how often a paused worker re-checks whether the pause
|
||||||
|
// window has closed.
|
||||||
|
const pauseCheckInterval = 60 * time.Second
|
||||||
|
|
||||||
// Options configures a Worker. Zero values are replaced with sensible defaults.
|
// Options configures a Worker. Zero values are replaced with sensible defaults.
|
||||||
type Options struct {
|
type Options struct {
|
||||||
QueueSize int // default 1000
|
QueueSize int // default 1000
|
||||||
Workers int // default 2
|
Workers int // default 2
|
||||||
Langs []string // default ["deu", "eng"]
|
Langs []string // default ["deu", "eng"]
|
||||||
Logger *slog.Logger
|
Logger *slog.Logger
|
||||||
|
// PausedHours optionally pauses processing during a local-time window
|
||||||
|
// [start, end) (PROJ-56). Wrap-around windows (e.g. [22, 6]) are supported.
|
||||||
|
// nil = never pause. The manual reprocess command leaves this nil on purpose.
|
||||||
|
PausedHours *[2]int
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewWorker constructs a worker that reads mails from store, runs OCR on
|
// NewWorker constructs a worker that reads mails from store, runs OCR on
|
||||||
@@ -66,9 +81,30 @@ func NewWorker(store *storage.Store, idxMgr index.TenantIndexer, opts Options) *
|
|||||||
done: make(chan struct{}),
|
done: make(chan struct{}),
|
||||||
workers: opts.Workers,
|
workers: opts.Workers,
|
||||||
langs: opts.Langs,
|
langs: opts.Langs,
|
||||||
|
pausedHours: opts.PausedHours,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// isPaused reports whether the OCR worker should currently hold off processing
|
||||||
|
// because the local time falls inside the configured pause window.
|
||||||
|
// Supports wrap-around windows where start > end (e.g. [22, 6]).
|
||||||
|
func (w *Worker) isPaused(now time.Time) bool {
|
||||||
|
if w.pausedHours == nil {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
start, end := w.pausedHours[0], w.pausedHours[1]
|
||||||
|
if start == end {
|
||||||
|
// Degenerate / no-op window — never pause.
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
h := now.Hour()
|
||||||
|
if start < end {
|
||||||
|
return h >= start && h < end
|
||||||
|
}
|
||||||
|
// Wrap-around window, e.g. [22, 6): paused 22,23,0,1,...,5.
|
||||||
|
return h >= start || h < end
|
||||||
|
}
|
||||||
|
|
||||||
// Submit enqueues a job. Drops with a warning if the queue is full so the
|
// Submit enqueues a job. Drops with a warning if the queue is full so the
|
||||||
// caller (mail intake) is never blocked.
|
// caller (mail intake) is never blocked.
|
||||||
func (w *Worker) Submit(mailID string, tenantID *int64) {
|
func (w *Worker) Submit(mailID string, tenantID *int64) {
|
||||||
@@ -108,6 +144,21 @@ func (w *Worker) Stop() {
|
|||||||
func (w *Worker) run(ctx context.Context, id int) {
|
func (w *Worker) run(ctx context.Context, id int) {
|
||||||
defer w.wg.Done()
|
defer w.wg.Done()
|
||||||
for {
|
for {
|
||||||
|
// PROJ-56: while inside the pause window, do NOT consume the queue.
|
||||||
|
// Jobs stay buffered in the channel (and as ocr_status='pending' in the
|
||||||
|
// DB), so nothing is lost. We re-check periodically and still react to
|
||||||
|
// shutdown / context cancellation immediately.
|
||||||
|
if w.isPaused(time.Now()) {
|
||||||
|
select {
|
||||||
|
case <-w.done:
|
||||||
|
return
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
case <-time.After(pauseCheckInterval):
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case job, ok := <-w.queue:
|
case job, ok := <-w.queue:
|
||||||
if !ok {
|
if !ok {
|
||||||
|
|||||||
Reference in New Issue
Block a user