pub/imap-copier
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reduce worker concurrency from 4 to 2

Browse Source 
Reduce idle read timeout from 120s to 60s

Add batching for IMAP metadata fetches
This commit is contained in:
vasyansk
2026-07-05 11:40:04 +07:00
parent b9cc7749c8
commit 2623bc8815
6 changed files with 125 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docker-compose.yml
+1 -1
View File
@@ -21,7 +21,7 @@ services:
AUTH_PASS: ${AUTH_PASS}
ENC_KEY: ${ENC_KEY}
SESSION_SECRET: ${SESSION_SECRET}
WORKER_CONCURRENCY: ${WORKER_CONCURRENCY:-4}
WORKER_CONCURRENCY: ${WORKER_CONCURRENCY:-2}
PPROF_ADDR: ${PPROF_ADDR:-:6060}
depends_on:
postgres:
internal/config/config.go
+1 -1
View File
@@ -25,7 +25,7 @@ func Load() (Config, error) {
AuthUser: os.Getenv("AUTH_USER"),
AuthPass: os.Getenv("AUTH_PASS"),
SessionSecret: []byte(os.Getenv("SESSION_SECRET")),
WorkerConcurrency: 4,
WorkerConcurrency: 2,
PprofAddr: os.Getenv("PPROF_ADDR"),
}
if v := os.Getenv("WORKER_CONCURRENCY"); v != "" {
internal/config/config_test.go
+2 -2
View File
@@ -31,7 +31,7 @@ func TestLoadDefaults(t *testing.T) {
if cfg.HTTPAddr != ":8080" {
t.Errorf("HTTPAddr = %q, want :8080", cfg.HTTPAddr)
}
if cfg.WorkerConcurrency != 4 {
t.Errorf("WorkerConcurrency = %d, want 4", cfg.WorkerConcurrency)
if cfg.WorkerConcurrency != 2 {
t.Errorf("WorkerConcurrency = %d, want 2", cfg.WorkerConcurrency)
}
}
internal/imapx/batch_test.go
+54
View File
@@ -0,0 +1,54 @@
package imapx
import (
"testing"
"github.com/emersion/go-imap/v2"
)
// metaBatches must tile the sequence 1..total into contiguous, non-overlapping
// windows of at most batchSize, covering every message exactly once. A single
// unbounded FETCH 1:* is what wedges large mailboxes; batching keeps each
// command short so the server stays responsive and ctx can be checked between
// windows.
func TestMetaBatches(t *testing.T) {
cases := []struct {
total, size uint32
want []imap.SeqRange
}{
{0, 1000, nil},
{1, 1000, []imap.SeqRange{{Start: 1, Stop: 1}}},
{1000, 1000, []imap.SeqRange{{Start: 1, Stop: 1000}}},
{1001, 1000, []imap.SeqRange{{Start: 1, Stop: 1000}, {Start: 1001, Stop: 1001}}},
{2500, 1000, []imap.SeqRange{{Start: 1, Stop: 1000}, {Start: 1001, Stop: 2000}, {Start: 2001, Stop: 2500}}},
}
for _, c := range cases {
got := metaBatches(c.total, c.size)
if len(got) != len(c.want) {
t.Fatalf("total=%d size=%d: got %d windows %v, want %d %v", c.total, c.size, len(got), got, len(c.want), c.want)
}
for i := range got {
if got[i] != c.want[i] {
t.Fatalf("total=%d size=%d window %d: got %+v want %+v", c.total, c.size, i, got[i], c.want[i])
}
}
}
}
// Every window must be within 1..total and the windows must be gap-free so no
// message is skipped or fetched twice.
func TestMetaBatchesCoverage(t *testing.T) {
const total, size = 4321, 1000
got := metaBatches(total, size)
if got[0].Start != 1 {
t.Fatalf("first window must start at 1, got %d", got[0].Start)
}
if last := got[len(got)-1]; last.Stop != total {
t.Fatalf("last window must stop at total=%d, got %d", total, last.Stop)
}
for i := 1; i < len(got); i++ {
if got[i].Start != got[i-1].Stop+1 {
t.Fatalf("gap/overlap between window %d (%+v) and %d (%+v)", i-1, got[i-1], i, got[i])
}
}
}
internal/imapx/conn.go
+1 -1
View File
@@ -11,7 +11,7 @@ import (
// live transfer never trips it — only a genuinely dead/mute socket does. This
// is what stops an account from wedging in "running" forever when a server
// accepts a command (e.g. a large FETCH) and then goes silent.
var idleReadTimeout = 120 * time.Second
var idleReadTimeout = 60 * time.Second
// idleConn wraps a net.Conn and arms a fresh read deadline before every Read.
// It lives beneath any TLS layer (the raw TCP conn), so the deadline governs
internal/imapx/copy.go
+66 -31
View File
@@ -36,6 +36,34 @@ type CopyResult struct {
Errors int
}
// metaScanBatch bounds how many messages one Pass-1 metadata FETCH covers. A
// single unbounded FETCH 1:* over a large mailbox keeps one command open for
// the entire scan; under parallel load the server can stop responding and,
// since go-imap has no per-command deadline, the worker wedges forever. Short
// windows keep each command brief so the server stays responsive and ctx is
// checked between windows.
const metaScanBatch = 1000
// metaBatches tiles 1..total into contiguous, non-overlapping windows of at
// most batchSize, covering every sequence number exactly once.
func metaBatches(total, batchSize uint32) []imap.SeqRange {
if total == 0 || batchSize == 0 {
return nil
}
var out []imap.SeqRange
for start := uint32(1); start <= total; start += batchSize {
stop := start + batchSize - 1
if stop > total {
stop = total
}
out = append(out, imap.SeqRange{Start: start, Stop: stop})
if stop == total {
break // guard against uint32 overflow when total is near max
}
}
return out
}
// CopyFolder streams messages from srcFolder on src to dstFolder on dst.
//
// The source folder is opened read-only (EXAMINE) and is never mutated:
@@ -71,45 +99,52 @@ func CopyFolder(ctx context.Context, src, dst *imapclient.Client, srcFolder, dst
internalDate time.Time
}
var todo []queued
metaSet := imap.SeqSet{imap.SeqRange{Start: 1, Stop: sel.NumMessages}}
fc := src.Fetch(metaSet, &imap.FetchOptions{
UID: true, Envelope: true, RFC822Size: true, Flags: true, InternalDate: true,
})
var scanned int64
for {
// Scan metadata in bounded windows instead of one FETCH 1:*, so each
// command is short (the server stays responsive) and ctx is checked on
// every window boundary — not just between messages of one giant command.
for _, win := range metaBatches(sel.NumMessages, metaScanBatch) {
if err := ctx.Err(); err != nil {
_ = fc.Close()
return res, err
}
msg := fc.Next()
if msg == nil {
break
}
buf, err := msg.Collect()
if err != nil {
res.Errors++
continue
}
scanned++
key := MessageKey(buf.Envelope, buf.RFC822Size)
already, err := deps.IsMigrated(key)
if err != nil {
res.Errors++
} else if already {
res.Skipped++
if deps.OnProgress != nil {
deps.OnProgress(res.Copied, res.Skipped)
fc := src.Fetch(imap.SeqSet{win}, &imap.FetchOptions{
UID: true, Envelope: true, RFC822Size: true, Flags: true, InternalDate: true,
})
for {
if err := ctx.Err(); err != nil {
_ = fc.Close()
return res, err
}
msg := fc.Next()
if msg == nil {
break
}
buf, err := msg.Collect()
if err != nil {
res.Errors++
continue
}
scanned++
key := MessageKey(buf.Envelope, buf.RFC822Size)
already, err := deps.IsMigrated(key)
if err != nil {
res.Errors++
} else if already {
res.Skipped++
if deps.OnProgress != nil {
deps.OnProgress(res.Copied, res.Skipped)
}
} else {
todo = append(todo, queued{uid: buf.UID, key: key, flags: buf.Flags, internalDate: buf.InternalDate})
}
if deps.OnScan != nil {
deps.OnScan(scanned, total)
}
} else {
todo = append(todo, queued{uid: buf.UID, key: key, flags: buf.Flags, internalDate: buf.InternalDate})
}
if deps.OnScan != nil {
deps.OnScan(scanned, total)
if err := fc.Close(); err != nil {
return res, fmt.Errorf("fetch meta %q: %w", srcFolder, err)
}
}
if err := fc.Close(); err != nil {
return res, fmt.Errorf("fetch meta %q: %w", srcFolder, err)
}
// Pass 2: fetch bodies for the queued (new) messages, one at a time.
for _, q := range todo {