scripte/archivmail
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(PROJ-58): Indexierung + OCR optional als Cron-Batch statt Dauerbetrieb

Browse Source 
Neues config.yml-Feld batch_mode (index/ocr, Default false = unverändertes
Verhalten). Bei batch_mode:true verarbeiten neue Cron-Jobs (index-pending,
ocr-reprocess) die Backlogs in größeren Abständen statt sofort bei jedem
Mail-Import, um Schreiblast auf der Festplatte zu glätten. Zeiten in
/etc/cron.d/archivmail frei anpassbar.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
sysops
2026-06-24 23:07:17 +02:00
co-authored by Claude Sonnet 4.6
parent 76655f78a2
commit fae274f930
10 changed files with 407 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/archivmail/cmd_import.go
+1
View File
@@ -300,6 +300,7 @@ Commands:
recompress Bestehende Mails nachträglich gzip-komprimieren
rethread Thread-IDs rückwirkend aus In-Reply-To/References befüllen
ocr-reprocess OCR für Anhänge nachholen (alle oder pro Mandant/Status)
index-pending Ungeindexte Mails nachindexieren (cron-fähig, PROJ-58 batch_mode)
update Auf neueste Version aktualisieren (führt update.sh aus)
status Healthcheck für DB, Manticore und Storage
version Version anzeigen
cmd/archivmail/cmd_index_pending.go
+157
View File
@@ -0,0 +1,157 @@
package main
import (
"context"
"flag"
"log/slog"
"os"
"strings"
"time"
"archivmail/config"
"archivmail/internal/index"
"archivmail/internal/storage"
"archivmail/pkg/mailparser"
)
// runIndexPending indexes all mails that have not yet been indexed
// (indexed_at IS NULL). It loads matching IDs from the DB, builds a
// MailDocument per mail, queues them on a TenantIndexWorker, waits for the
// worker to drain, then exits. Designed to be driven by cron when
// index.batch_mode is enabled (PROJ-58).
//
// Usage:
//
// archivmail index-pending --config /etc/archivmail/config.yml
// archivmail index-pending --limit 500
func runIndexPending(args []string) {
fs := flag.NewFlagSet("index-pending", flag.ExitOnError)
configPath := fs.String("config", "/etc/archivmail/config.yml", "path to config file")
limitFlag := fs.Int("limit", 0, "max number of mails to index (0 = no limit)")
fs.Parse(args)
logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
cfg, err := config.Load(*configPath)
if err != nil {
logger.Error("failed to load config", "err", err)
os.Exit(1)
}
storeCfg := storage.Config{
Dir: cfg.Storage.StorePath,
Keyfile: cfg.Storage.Keyfile,
DSN: cfg.Database.DSN(),
CompressEnabled: cfg.Storage.Compress,
}
mailStore, err := storage.New(storeCfg)
if err != nil {
logger.Error("storage init failed", "err", err)
os.Exit(1)
}
defer mailStore.Close()
indexBackend := cfg.Index.Backend
if indexBackend == "" {
indexBackend = "manticore"
}
batchSize := cfg.Index.BatchSize
if batchSize <= 0 {
batchSize = 100
}
var idxMgr index.TenantIndexer
if indexBackend == "manticore" {
dsn := cfg.Index.ManticoreDSN
if dsn == "" {
dsn = "manticore@tcp(127.0.0.1:9306)/?charset=utf8mb4"
}
m, err := index.NewManticoreTenantManager(dsn)
if err != nil {
logger.Error("manticore init failed", "err", err)
os.Exit(1)
}
idxMgr = m
} else {
m, err := index.NewTenantIndexManager(cfg.Index.Path, batchSize, indexBackend)
if err != nil {
logger.Error("index manager init failed", "err", err)
os.Exit(1)
}
idxMgr = m
}
defer idxMgr.Close()
ctx := context.Background()
mails, err := mailStore.GetUnindexedMails(ctx, *limitFlag)
if err != nil {
logger.Error("failed to list unindexed mails", "err", err)
os.Exit(1)
}
logger.Info("index-pending: starting", "count", len(mails), "limit", *limitFlag)
if len(mails) == 0 {
return
}
// Queue size needs to fit the entire batch so Submit never drops.
qSize := len(mails) + 16
worker := index.NewTenantWorker(idxMgr, qSize, logger)
worker.Start()
// Periodic progress while waiting for the queue to drain.
tick := time.NewTicker(5 * time.Second)
defer tick.Stop()
go func() {
for range tick.C {
logger.Info("index-pending: progress", "queue_remaining", worker.QueueLen())
}
}()
submitted := 0
for _, m := range mails {
raw, err := mailStore.Load(m.ID)
if err != nil {
logger.Warn("index-pending: load failed", "id", m.ID, "err", err)
continue
}
pm, err := mailparser.Parse(raw)
if err != nil {
logger.Warn("index-pending: parse failed, skipping", "id", m.ID, "err", err)
continue
}
var attachNames []string
for _, a := range pm.Attachments {
if a.Filename != "" {
attachNames = append(attachNames, a.Filename)
}
}
doc := index.MailDocument{
ID: m.ID,
From: pm.From,
To: strings.Join(pm.To, ", "),
CC: strings.Join(pm.CC, ", "),
Subject: pm.Subject,
Body: pm.TextBody,
AttachNames: strings.Join(attachNames, " "),
HasAttachment: len(pm.Attachments) > 0,
Date: pm.Date,
Size: int64(len(raw)),
TenantID: m.TenantID,
}
worker.Submit(doc)
// Mark as indexed in DB so subsequent runs skip it.
if err := mailStore.SetIndexedAt(ctx, m.ID); err != nil {
logger.Warn("index-pending: set indexed_at failed", "id", m.ID, "err", err)
}
submitted++
}
worker.Stop() // waits for the queue to drain
logger.Info("index-pending: complete", "submitted", submitted)
}
cmd/archivmail/main.go
+61 -19
View File
@@ -70,6 +70,9 @@ func main() {
case "ocr-reprocess":
runOCRReprocess(os.Args[2:])
return
case "index-pending":
runIndexPending(os.Args[2:])
return
case "update":
runUpdate(os.Args[2:])
return
@@ -183,8 +186,14 @@ func main() {
asyncQueueSize = 1000
}
tenantWorker := index.NewTenantWorker(idxMgr, asyncQueueSize, logger)
tenantWorker.Start()
defer tenantWorker.Stop()
// PROJ-58: in batch mode the continuous index worker is not started; the
// `index-pending` cron job indexes mails (indexed_at IS NULL) instead.
if !cfg.Index.BatchMode {
tenantWorker.Start()
defer tenantWorker.Stop()
} else {
logger.Info("index worker: batch mode enabled — continuous worker not started (use 'archivmail index-pending' via cron)")
}
// PROJ-35: OCR-Worker — extracts text from PDF/image attachments and feeds
// it back into the per-tenant Manticore index. Non-blocking submit so the
@@ -202,8 +211,14 @@ func main() {
logger.Info("ocr worker: pause window configured",
"from_hour", cfg.OCR.PausedHours[0], "to_hour", cfg.OCR.PausedHours[1])
}
ocrWorker.Start(context.Background())
defer ocrWorker.Stop()
// PROJ-58: in batch mode the continuous OCR worker is not started; the
// `ocr-reprocess --status pending` cron job processes the backlog instead.
if !cfg.OCR.BatchMode {
ocrWorker.Start(context.Background())
defer ocrWorker.Stop()
} else {
logger.Info("ocr worker: batch mode enabled — continuous worker not started (use 'archivmail ocr-reprocess' via cron)")
}
if !ocr.IsAvailable() {
ts := ocr.CheckTools()
logger.Warn("ocr tools not fully available — install tesseract-ocr + poppler-utils for full OCR support",
@@ -216,7 +231,9 @@ func main() {
// pending mails as currently fit, so nothing is dropped.
// The worker updates ocr_status to done/failed/skipped, so subsequent
// queries only return genuinely outstanding jobs.
go func() {
// PROJ-58: skipped in batch mode — the cron job drains the backlog instead.
if !cfg.OCR.BatchMode {
go func() {
ctx := context.Background()
queueCap := 1000 // matches ocr.Options.QueueSize above
processed := 0
@@ -250,8 +267,9 @@ func main() {
processed += len(pending)
logger.Info("ocr boot-resume: enqueued batch",
"batch", len(pending), "total_so_far", processed)
}
}()
}
}()
}
// User store
users, err := userstore.New(cfg.Database.DSN())
@@ -385,7 +403,7 @@ func main() {
smtpDaemon.SetIndexCallback(func(raw []byte, id string) {
// Look up the tenant_id for this email from DB metadata.
tenantID, _ := mailStore.GetTenantForMail(context.Background(), id)
submitToWorker(tenantWorker, mailStore, raw, id, tenantID, logger, ocrWorker)
submitToWorker(tenantWorker, mailStore, raw, id, tenantID, logger, ocrWorker, cfg.Index.BatchMode, cfg.OCR.BatchMode)
})
// Wire tenant routing into SMTP daemon
if cfg.SMTP.TenantRouting == "domain" {
@@ -435,7 +453,8 @@ func main() {
imapImp := imapstore.NewImporter(imapSt, mailStore, idxMgr, logger)
// PROJ-44: trigger OCR for IMAP-imported mails — without this every
// IMAP delivery would remain in ocr_status='pending' forever.
if ocrWorker != nil {
// PROJ-58: skipped in OCR batch mode (cron job drains the backlog).
if ocrWorker != nil && !cfg.OCR.BatchMode {
imapImp.SetOCRSubmit(func(mailID string, tenantID *int64) {
ocrWorker.Submit(mailID, tenantID)
})
@@ -456,15 +475,26 @@ func main() {
defer pop3St.Close()
pop3Imp := pop3store.NewImporter(pop3St, mailStore, idxMgr, logger)
// PROJ-44: same OCR hook as the IMAP importer above.
if ocrWorker != nil {
// PROJ-58: skipped in OCR batch mode (cron job drains the backlog).
if ocrWorker != nil && !cfg.OCR.BatchMode {
pop3Imp.SetOCRSubmit(func(mailID string, tenantID *int64) {
ocrWorker.Submit(mailID, tenantID)
})
}
srv.SetPop3(pop3St, pop3Imp)
// Backfill in background: migrate existing files into DB metadata + re-index
go runBackfill(context.Background(), mailStore, idx, tenantWorker, logger, ocrWorker)
// Backfill in background: migrate existing files into DB metadata + re-index.
// PROJ-58: skipped when index.batch_mode is set — the `index-pending` cron
// job indexes outstanding mails (indexed_at IS NULL) instead. When OCR is
// also in batch mode, hand a nil ocrWorker to runBackfill so it does not
// submit OCR jobs to the (unstarted) in-memory queue.
if !cfg.Index.BatchMode {
backfillOCR := ocrWorker
if cfg.OCR.BatchMode {
backfillOCR = nil
}
go runBackfill(context.Background(), mailStore, idx, tenantWorker, logger, backfillOCR)
}
// Background integrity verification — runs every 5 minutes
go runIntegrityCheck(context.Background(), mailStore, logger)
@@ -522,7 +552,11 @@ func reloadOCRPauseWindow(configPath string, ocrWorker *ocr.Worker, logger *slog
// tenantID may be nil for global context.
// If ocrWorker is non-nil and the mail has attachments, an OCR job is also
// queued (non-blocking).
func submitToWorker(worker *index.TenantIndexWorker, store *storage.Store, raw []byte, id string, tenantID *int64, logger *slog.Logger, ocrWorker *ocr.Worker) {
// indexBatchMode / ocrBatchMode (PROJ-58): when set, the corresponding
// in-memory submit is skipped so the unstarted batch-mode worker queue does
// not fill up and log spurious "queue full" warnings. The mail still gets its
// indexed_at / ocr_status state so the cron batch jobs pick it up.
func submitToWorker(worker *index.TenantIndexWorker, store *storage.Store, raw []byte, id string, tenantID *int64, logger *slog.Logger, ocrWorker *ocr.Worker, indexBatchMode, ocrBatchMode bool) {
pm, err := mailparser.Parse(raw)
if err != nil {
logger.Warn("index: parse failed, skipping indexing", "id", id, "err", err)
@@ -550,15 +584,21 @@ func submitToWorker(worker *index.TenantIndexWorker, store *storage.Store, raw [
TenantID: tenantID,
}
worker.Submit(doc)
// PROJ-58: in index batch mode the continuous worker is not running; leave
// the mail with indexed_at IS NULL so `index-pending` picks it up via cron.
if !indexBatchMode {
worker.Submit(doc)
// Mark as indexed in DB
if err := store.SetIndexedAt(context.Background(), id); err != nil {
logger.Warn("index: set indexed_at failed", "id", id, "err", err)
// Mark as indexed in DB
if err := store.SetIndexedAt(context.Background(), id); err != nil {
logger.Warn("index: set indexed_at failed", "id", id, "err", err)
}
}
// PROJ-35: hand off to OCR worker for asynchronous attachment processing.
if ocrWorker != nil && len(pm.Attachments) > 0 {
// PROJ-58: skipped in OCR batch mode — the mail stays ocr_status='pending'
// (set at storage time) and is processed by the ocr-reprocess cron job.
if !ocrBatchMode && ocrWorker != nil && len(pm.Attachments) > 0 {
ocrWorker.Submit(id, tenantID)
}
}
@@ -612,7 +652,9 @@ func runBackfill(ctx context.Context, store *storage.Store, idx index.Indexer, w
if !alreadyIndexed {
needIndex++
tenantID, _ := store.GetTenantForMail(ctx, id)
submitToWorker(worker, store, raw, id, tenantID, logger, ocrWorker)
// runBackfill only runs when index.batch_mode is off; the OCR
// batch case is handled by passing a nil ocrWorker from the caller.
submitToWorker(worker, store, raw, id, tenantID, logger, ocrWorker, false, false)
}
if count%100 == 0 {