sysopsandClaude Sonnet 5
798cb2817c
Backend: recover() in allen langlebigen Goroutinen (neues internal/safego- Paket), MIME-Multipart-Tiefenlimit gegen Stack-Overflow, IMAP-Zeilenlängen- und FETCH-Result-Limits gegen OOM, MBOX-Buffer-Aliasing-Bug (Datenkorruption beim Import), ungeprüfte Type Assertions abgesichert, Data Race im API-Key-Rate-Limiter behoben, SMTP-Session-Panic führt jetzt zu 451-Retry statt Prozessabsturz. Performance: O(n²)-String-Concat in Mailparser und IMAP-Parser durch strings.Builder ersetzt. Frontend: Error Boundaries für Root und Mail-Detailansicht ergänzt (gab es vorher nicht), zahlreiche Guards gegen nil-Slices aus dem Backend-JSON die sonst .map()/.length-Crashes/White-Screens auslösten, defekte JSON-Antworten in api/core.ts abgefangen, zwei React-Key-Bugs bei löschbaren Listen korrigiert. Verifiziert auf 192.168.1.132: Build und Tests der geänderten Pakete fehlerfrei, keine Regressionen gegenüber vorbestehendem Stand. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019j28kGcaJAhBnrYX34hGdt
105 lines
2.6 KiB
Go
105 lines
2.6 KiB
Go
package index
|
|
|
|
import (
|
|
"fmt"
|
|
"log/slog"
|
|
"sync"
|
|
)
|
|
|
|
// TenantIndexWorker processes MailDocument indexing requests asynchronously,
|
|
// routing each document to the correct per-tenant index via TenantIndexer.
|
|
type TenantIndexWorker struct {
|
|
mgr TenantIndexer
|
|
queue chan MailDocument
|
|
done chan struct{}
|
|
wg sync.WaitGroup
|
|
logger *slog.Logger
|
|
}
|
|
|
|
// NewTenantWorker creates a new TenantIndexWorker with the given queue capacity.
|
|
func NewTenantWorker(mgr TenantIndexer, queueSize int, logger *slog.Logger) *TenantIndexWorker {
|
|
if queueSize <= 0 {
|
|
queueSize = 1000
|
|
}
|
|
return &TenantIndexWorker{
|
|
mgr: mgr,
|
|
queue: make(chan MailDocument, queueSize),
|
|
done: make(chan struct{}),
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
// Submit enqueues a document for background indexing. If the queue is full,
|
|
// the document is dropped and a warning is logged.
|
|
func (w *TenantIndexWorker) Submit(doc MailDocument) {
|
|
select {
|
|
case w.queue <- doc:
|
|
// queued
|
|
default:
|
|
w.logger.Warn("tenant index worker: queue full, dropping document", "id", doc.ID)
|
|
}
|
|
}
|
|
|
|
// Start launches the background goroutine that processes the queue.
|
|
func (w *TenantIndexWorker) Start() {
|
|
w.wg.Add(1)
|
|
go func() {
|
|
defer w.wg.Done()
|
|
w.logger.Info("tenant index worker: started", "queue_size", cap(w.queue))
|
|
for {
|
|
select {
|
|
case doc, ok := <-w.queue:
|
|
if !ok {
|
|
return
|
|
}
|
|
w.indexDoc(doc)
|
|
case <-w.done:
|
|
// Drain remaining items in the queue before exiting.
|
|
for {
|
|
select {
|
|
case doc, ok := <-w.queue:
|
|
if !ok {
|
|
return
|
|
}
|
|
w.indexDoc(doc)
|
|
default:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
// Stop signals the worker to drain remaining items and stop.
|
|
func (w *TenantIndexWorker) Stop() {
|
|
close(w.done)
|
|
w.wg.Wait()
|
|
w.logger.Info("tenant index worker: stopped")
|
|
}
|
|
|
|
// QueueLen returns the current number of items waiting in the queue.
|
|
func (w *TenantIndexWorker) QueueLen() int {
|
|
return len(w.queue)
|
|
}
|
|
|
|
func (w *TenantIndexWorker) indexDoc(doc MailDocument) {
|
|
// A panic while indexing a single document must not kill the long-running
|
|
// worker goroutine (and with it the whole process).
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
w.logger.Error("tenant index worker: recovered from panic",
|
|
"id", doc.ID, "tenant_id", doc.TenantID, "panic", fmt.Sprintf("%v", r))
|
|
}
|
|
}()
|
|
|
|
idx := w.mgr.ForTenant(doc.TenantID)
|
|
if idx == nil {
|
|
w.logger.Error("tenant index worker: no indexer for tenant", "id", doc.ID, "tenant_id", doc.TenantID)
|
|
return
|
|
}
|
|
if err := idx.IndexSync(doc); err != nil {
|
|
w.logger.Error("tenant index worker: index failed", "id", doc.ID, "tenant_id", doc.TenantID, "err", err)
|
|
}
|
|
}
|