diff --git a/internal/imapx/keepalive.go b/internal/imapx/keepalive.go new file mode 100644 index 0000000..4e41f24 --- /dev/null +++ b/internal/imapx/keepalive.go @@ -0,0 +1,47 @@ +package imapx + +import ( + "context" + "time" + + "github.com/emersion/go-imap/v2/imapclient" +) + +// KeepaliveInterval is how often an otherwise-idle IMAP connection is pinged +// with NOOP so the server does not drop it. +// +// The destination connection sits completely idle for the entire duration of +// the source-side metadata scan (Pass 1 of CopyFolder), which on a large +// mailbox runs for many minutes across all folders. With no traffic, the +// server closes the idle connection; go-imap's reader then tears the client +// down, and every subsequent APPEND fails with "use of closed network +// connection" โ€” aborting each folder and copying nothing. A periodic NOOP +// keeps the connection warm. 60s is well under the idle timeout of any common +// IMAP server. +const KeepaliveInterval = 60 * time.Second + +// Keepalive pings c with a NOOP every interval until ctx is cancelled, keeping +// an idle connection from being dropped by the server. It is meant to run in +// its own goroutine. +// +// It is safe to run concurrently with other commands on c: go-imap serializes +// command submission and supports multiple in-flight commands over a single +// connection (RFC 9051 ยง5.5 pipelining). A NOOP issued while another command is +// in flight simply queues behind it and completes when the server responds. +// +// Keepalive returns when ctx is done or when a NOOP fails โ€” a failed NOOP means +// the connection is already gone, so there is nothing left to keep alive. +func Keepalive(ctx context.Context, c *imapclient.Client, interval time.Duration) { + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + if err := c.Noop().Wait(); err != nil { + return + } + } + } +} diff --git a/internal/imapx/keepalive_test.go b/internal/imapx/keepalive_test.go new file mode 100644 index 0000000..f9ef132 --- /dev/null +++ b/internal/imapx/keepalive_test.go @@ -0,0 +1,89 @@ +package imapx + +import ( + "context" + "testing" + "time" +) + +// TestKeepaliveReturnsOnContextCancel proves Keepalive is a well-behaved +// goroutine: it exits promptly when its context is cancelled instead of +// leaking. +func TestKeepaliveReturnsOnContextCancel(t *testing.T) { + ep := testEP(t) + ctx := context.Background() + + c, err := Connect(ctx, ep) + if err != nil { + t.Fatal(err) + } + defer func() { _ = c.Logout().Wait() }() + if err := c.Login("ka1@localhost", "p").Wait(); err != nil { + t.Fatal(err) + } + + kctx, cancel := context.WithCancel(ctx) + done := make(chan struct{}) + go func() { Keepalive(kctx, c, 10*time.Millisecond); close(done) }() + + // Let a few NOOPs fire, then cancel and require a prompt return. + time.Sleep(50 * time.Millisecond) + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Keepalive did not return within 2s of context cancel") + } +} + +// TestKeepaliveDoesNotDisruptCopy runs Keepalive on the destination connection +// at an aggressive interval while CopyFolder is APPENDing to it, proving the +// concurrent NOOPs do not corrupt in-flight commands (the real risk of pinging +// a connection that is also being used for real work). +func TestKeepaliveDoesNotDisruptCopy(t *testing.T) { + ep := testEP(t) + ctx := context.Background() + + const n = 8 + seedInbox(t, ep, "kasrc@localhost", "p", n) + + src, err := Connect(ctx, ep) + if err != nil { + t.Fatal(err) + } + defer func() { _ = src.Logout().Wait() }() + if err := src.Login("kasrc@localhost", "p").Wait(); err != nil { + t.Fatal(err) + } + + dst, err := Connect(ctx, ep) + if err != nil { + t.Fatal(err) + } + defer func() { _ = dst.Logout().Wait() }() + if err := dst.Login("kadst@localhost", "p").Wait(); err != nil { + t.Fatal(err) + } + + kctx, cancel := context.WithCancel(ctx) + defer cancel() + go Keepalive(kctx, dst, 1*time.Millisecond) + + seen := map[string]bool{} + deps := CopyDeps{ + IsMigrated: func(k string) (bool, error) { return seen[k], nil }, + MarkMigrated: func(_, k string) error { seen[k] = true; return nil }, + OnProgress: func(_, _ int) {}, + } + + r, err := CopyFolder(kctx, src, dst, "INBOX", "INBOX", deps) + if err != nil { + t.Fatalf("CopyFolder with concurrent keepalive: %v", err) + } + if r.Copied != n { + t.Fatalf("copied=%d want %d", r.Copied, n) + } + if r.Errors != 0 { + t.Fatalf("errors=%d want 0", r.Errors) + } +} diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 96cfff1..7038112 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -354,6 +354,15 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in _ = dst.Close() }() + // Keep the destination connection warm. It sits idle for the whole + // source-side metadata scan (Pass 1 of CopyFolder), which on a large + // mailbox runs for minutes; without traffic the server drops it and every + // subsequent APPEND fails with "use of closed network connection", copying + // nothing. Periodic NOOPs prevent that. Only dst needs it โ€” src is + // continuously busy scanning/fetching. Bound to actx so it stops with the + // account. + go imapx.Keepalive(actx, dst, imapx.KeepaliveInterval) + // Progress watchdog: track the last time we saw scan/copy activity; if it // goes quiet for stallTimeout, cancel the account so the connections close // and this worker unwinds (it would otherwise block forever on a silent