fix dtt

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
+			}
+		}
+	}
+}

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)
+	}
+}

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