diff --git a/internal/imapx/account.go b/internal/imapx/account.go index b38ccdc..a8846da 100644 --- a/internal/imapx/account.go +++ b/internal/imapx/account.go @@ -4,13 +4,12 @@ import ( "context" "github.com/emersion/go-imap/v2" - "github.com/emersion/go-imap/v2/imapclient" ) // FolderMessageCount opens a folder read-only (EXAMINE) and returns how many // messages it holds — used to plan an accurate overall progress total before // copying begins. It does not fetch any message bodies. -func FolderMessageCount(c *imapclient.Client, folder string) (int64, error) { +func FolderMessageCount(c *Client, folder string) (int64, error) { sel, err := c.Select(folder, &imap.SelectOptions{ReadOnly: true}).Wait() if err != nil { return 0, err @@ -19,7 +18,7 @@ func FolderMessageCount(c *imapclient.Client, folder string) (int64, error) { } // ListFolders returns the mailbox names visible on an already-connected, logged-in client. -func ListFolders(c *imapclient.Client) ([]string, error) { +func ListFolders(c *Client) ([]string, error) { mboxes, err := c.List("", "*", nil).Collect() if err != nil { return nil, err diff --git a/internal/imapx/copy.go b/internal/imapx/copy.go index 4aae57e..f79fda7 100644 --- a/internal/imapx/copy.go +++ b/internal/imapx/copy.go @@ -8,6 +8,7 @@ import ( "io" "log/slog" "net" + "os" "strings" "time" @@ -42,8 +43,31 @@ type CopyDeps struct { // in-body signal the watchdog can't tell a slow-but-live transfer from a // wedged connection and cancels a healthy copy. May be nil. OnActivity func() + // ReconnectSrc dials and logs in a FRESH source client, returning it. It is + // called when a body read times out (server under-delivered a literal), + // which leaves the current src connection desynced and unusable. CopyFolder + // swaps to the returned client, re-EXAMINEs the folder, and resumes. The + // implementation is expected to also update any external reference to the + // live src client (e.g. so a cancel path closes the right connection). If + // nil, a body-read timeout aborts the folder instead of recovering. + ReconnectSrc func() (*Client, error) } +// ErrBodyTimeout means a message body did not finish transferring within the +// idle deadline — the server stopped sending mid-literal. It is almost always a +// server announcing a BODY[] literal larger than the bytes it actually sends, +// which makes go-imap wait forever for bytes that never come. Distinct from a +// closed connection so the caller can skip just this one message and resume. +var ErrBodyTimeout = errors.New("message body read timed out") + +// bodyIdleTimeout bounds how long a body read may go with NO bytes arriving +// before it is abandoned. It is generous enough for legitimately slow servers +// (even ~15 KB/s links keep bytes flowing far more often than this) yet well +// under the orchestrator's multi-minute stall watchdog, so an under-delivered +// literal is caught quickly and locally instead of stalling the whole account. +// A var (not const) so tests can shorten it. +var bodyIdleTimeout = 30 * time.Second + // CopyResult summarizes the outcome of one CopyFolder run. type CopyResult struct { Copied int @@ -95,7 +119,7 @@ func metaBatches(total, batchSize uint32) []imap.SeqRange { // held in memory only for the duration of a single FETCH->APPEND and is // never written to disk. Messages already migrated (per deps.IsMigrated) // are skipped without re-fetching their bodies. -func CopyFolder(ctx context.Context, src, dst *imapclient.Client, srcFolder, dstFolder string, deps CopyDeps) (CopyResult, error) { +func CopyFolder(ctx context.Context, src, dst *Client, srcFolder, dstFolder string, deps CopyDeps) (CopyResult, error) { var res CopyResult sel, err := src.Select(srcFolder, &imap.SelectOptions{ReadOnly: true}).Wait() @@ -190,6 +214,25 @@ func CopyFolder(ctx context.Context, src, dst *imapclient.Client, srcFolder, dst if err := streamOne(src, dst, dstFolder, q.uid, q.flags, q.internalDate, deps.OnActivity); err != nil { res.Errors++ reportErr(msgRef(q.uid, q.subject), "copy message: "+err.Error()) + // A body-read timeout means the server under-delivered this message's + // literal; the src connection is now desynced. Mark the message + // migrated so this and future runs skip it (it is un-fetchable via a + // conforming client), then reconnect src and resume the folder with + // the remaining queued messages. + if errors.Is(err, ErrBodyTimeout) && deps.ReconnectSrc != nil { + if merr := deps.MarkMigrated(dstFolder, q.key); merr != nil { + reportErr(msgRef(q.uid, q.subject), "mark skipped: "+merr.Error()) + } + newSrc, rerr := deps.ReconnectSrc() + if rerr != nil { + return res, fmt.Errorf("reconnect src after body timeout in %q: %w", srcFolder, rerr) + } + src = newSrc + if _, serr := src.Select(srcFolder, &imap.SelectOptions{ReadOnly: true}).Wait(); serr != nil { + return res, fmt.Errorf("re-examine %q after reconnect: %w", srcFolder, serr) + } + continue + } // A closed/broken connection won't recover: every remaining APPEND // would fail identically. Abort the folder instead of logging // thousands of the same error; a re-run resumes via dedup. @@ -218,17 +261,26 @@ func CopyFolder(ctx context.Context, src, dst *imapclient.Client, srcFolder, dst // evidence of which side and which message is the culprit. const slowMessage = 20 * time.Second -// touchReader wraps a body Read stream and pings onActivity on every non-empty -// read, so a long FETCH keeps the stall watchdog fed byte-by-byte. -type touchReader struct { - r io.Reader - on func() +// deadlineReader arms an idle read deadline on the source socket before every +// read of a message body, so a server that stops sending mid-literal — having +// announced a larger BODY[] size than it actually delivers — trips the deadline +// instead of blocking go-imap forever waiting for bytes that never arrive. It +// also pings onActivity as bytes arrive, feeding the orchestrator's stall +// watchdog. The deadline is refreshed on each read, so it bounds IDLE time +// (no bytes) rather than total transfer time — a legitimately slow but steady +// download never trips it. +type deadlineReader struct { + c *Client + r io.Reader + idle time.Duration + on func() } -func (t touchReader) Read(p []byte) (int, error) { - n, err := t.r.Read(p) - if n > 0 && t.on != nil { - t.on() +func (d deadlineReader) Read(p []byte) (int, error) { + _ = d.c.SetReadDeadline(time.Now().Add(d.idle)) + n, err := d.r.Read(p) + if n > 0 && d.on != nil { + d.on() } return n, err } @@ -252,7 +304,11 @@ func (t touchWriter) Write(p []byte) (int, error) { // spooling to disk. The body is buffered in RAM only for the duration of // this single FETCH->APPEND round trip. onActivity (may be nil) fires as bytes // move in either direction, feeding the orchestrator's stall watchdog. -func streamOne(src, dst *imapclient.Client, dstFolder string, uid imap.UID, flags []imap.Flag, internalDate time.Time, onActivity func()) error { +// +// The body read is guarded by an idle deadline on src: if the server goes +// silent mid-literal, streamOne returns ErrBodyTimeout rather than hanging, so +// CopyFolder can skip the message and reconnect. +func streamOne(src, dst *Client, dstFolder string, uid imap.UID, flags []imap.Flag, internalDate time.Time, onActivity func()) error { bodySection := &imap.FetchItemBodySection{} fetchStart := time.Now() fetchCmd := src.Fetch(imap.UIDSetNum(uid), &imap.FetchOptions{ @@ -265,19 +321,25 @@ func streamOne(src, dst *imapclient.Client, dstFolder string, uid imap.UID, flag return fmt.Errorf("no message for uid %v", uid) } var body []byte + var readErr error for { item := msg.Next() if item == nil { break } if d, ok := item.(imapclient.FetchItemDataBodySection); ok { - b, err := io.ReadAll(touchReader{r: d.Literal, on: onActivity}) - if err != nil { - return err - } - body = b + body, readErr = io.ReadAll(deadlineReader{c: src, r: d.Literal, idle: bodyIdleTimeout, on: onActivity}) } } + // Clear the deadline before any further I/O on src (fetchCmd.Close reads the + // command's completion off the same socket). + _ = src.SetReadDeadline(time.Time{}) + if readErr != nil { + if errors.Is(readErr, os.ErrDeadlineExceeded) { + return fmt.Errorf("%w: uid %v (server sent fewer bytes than the announced literal)", ErrBodyTimeout, uid) + } + return readErr + } if err := fetchCmd.Close(); err != nil { return err } diff --git a/internal/imapx/dial.go b/internal/imapx/dial.go index e0c4c86..517268d 100644 --- a/internal/imapx/dial.go +++ b/internal/imapx/dial.go @@ -18,6 +18,29 @@ type Endpoint struct { func (e Endpoint) addr() string { return fmt.Sprintf("%s:%d", e.Host, e.Port) } +// Client wraps an imapclient.Client together with the raw network connection it +// runs over. The embedded *imapclient.Client provides the full IMAP API; the +// retained conn lets callers impose a read deadline on the socket for the +// duration of a body transfer. +// +// This defends against servers that announce a BODY[] literal LARGER than the +// bytes they actually send (a protocol violation observed on some webmail +// servers). go-imap reads a literal strictly by its announced size, so a short +// literal makes it block forever waiting for bytes that never arrive. A +// deadline around the body read turns that infinite hang into a timeout the +// copier can recover from. +type Client struct { + *imapclient.Client + conn net.Conn +} + +// SetReadDeadline sets (or, with the zero time, clears) a deadline on the +// underlying socket. Used to bound a single body read; always cleared again +// once the read completes so it never affects idle periods. +func (c *Client) SetReadDeadline(t time.Time) error { + return c.conn.SetReadDeadline(t) +} + // dialTimeout bounds establishing the TCP connection. const dialTimeout = 30 * time.Second @@ -27,7 +50,7 @@ const dialTimeout = 30 * time.Second // during a long src scan) from one stuck mid-response, and would wrongly close // idle connections. Stall detection is done at the orchestrator level via a // progress watchdog; go-imap's own per-command timeouts bound active commands. -func dialOnce(ctx context.Context, ep Endpoint) (*imapclient.Client, error) { +func dialOnce(ctx context.Context, ep Endpoint) (*Client, error) { d := &net.Dialer{Timeout: dialTimeout} raw, err := d.DialContext(ctx, "tcp", ep.addr()) if err != nil { @@ -42,16 +65,27 @@ func dialOnce(ctx context.Context, ep Endpoint) (*imapclient.Client, error) { _ = raw.Close() return nil, err } - return waitGreeting(imapclient.New(tlsConn, nil)) + c, err := waitGreeting(imapclient.New(tlsConn, nil)) + if err != nil { + return nil, err + } + return &Client{Client: c, conn: tlsConn}, nil case "starttls": + // Deadline goes on the raw TCP conn: it sits beneath the TLS layer that + // NewStartTLS negotiates, and a TCP read deadline still interrupts the + // TLS read above it. opts := &imapclient.Options{TLSConfig: &tls.Config{ServerName: ep.Host}} c, err := imapclient.NewStartTLS(raw, opts) if err != nil { return nil, err } - return c, nil + return &Client{Client: c, conn: raw}, nil case "plain": - return waitGreeting(imapclient.New(raw, nil)) + c, err := waitGreeting(imapclient.New(raw, nil)) + if err != nil { + return nil, err + } + return &Client{Client: c, conn: raw}, nil default: _ = raw.Close() return nil, fmt.Errorf("unknown tls_mode %q", ep.TLSMode) @@ -69,7 +103,7 @@ func waitGreeting(c *imapclient.Client) (*imapclient.Client, error) { return c, nil } -func Connect(ctx context.Context, ep Endpoint) (*imapclient.Client, error) { +func Connect(ctx context.Context, ep Endpoint) (*Client, error) { const attempts = 3 var lastErr error for i := 0; i < attempts; i++ { diff --git a/internal/imapx/keepalive.go b/internal/imapx/keepalive.go index 4e41f24..f398130 100644 --- a/internal/imapx/keepalive.go +++ b/internal/imapx/keepalive.go @@ -3,8 +3,6 @@ package imapx import ( "context" "time" - - "github.com/emersion/go-imap/v2/imapclient" ) // KeepaliveInterval is how often an otherwise-idle IMAP connection is pinged @@ -31,7 +29,7 @@ const KeepaliveInterval = 60 * time.Second // // 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) { +func Keepalive(ctx context.Context, c *Client, interval time.Duration) { t := time.NewTicker(interval) defer t.Stop() for { diff --git a/internal/imapx/underflow_test.go b/internal/imapx/underflow_test.go new file mode 100644 index 0000000..6a88027 --- /dev/null +++ b/internal/imapx/underflow_test.go @@ -0,0 +1,218 @@ +package imapx + +import ( + "bufio" + "context" + "errors" + "fmt" + "net" + "strings" + "sync" + "testing" + "time" + + "github.com/emersion/go-imap/v2" +) + +// underflowServer is a minimal IMAP server that reproduces the amega.kz bug: it +// answers a BODY[] FETCH by announcing a literal LARGER than the bytes it then +// sends, and afterwards goes silent — exactly what makes go-imap's strict +// literal reader block forever. LOGIN/EXAMINE and the Pass-1 metadata FETCH are +// answered normally so a full CopyFolder can reach the poisoned message. +// +// It serves one connection per accept and keeps accepting, so a reconnect gets +// a fresh, well-behaved session (its second EXAMINE reports zero messages, so +// the resumed folder simply finishes). +type underflowServer struct { + ln net.Listener + mu sync.Mutex + accepts int // how many connections have been accepted + stop chan struct{} +} + +func newUnderflowServer(t *testing.T) *underflowServer { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + s := &underflowServer{ln: ln, stop: make(chan struct{})} + go s.serve() + return s +} + +func (s *underflowServer) addr() Endpoint { + a := s.ln.Addr().(*net.TCPAddr) + return Endpoint{Host: "127.0.0.1", Port: a.Port, TLSMode: "plain"} +} + +func (s *underflowServer) close() { close(s.stop); _ = s.ln.Close() } + +func (s *underflowServer) serve() { + for { + conn, err := s.ln.Accept() + if err != nil { + return + } + s.mu.Lock() + s.accepts++ + first := s.accepts == 1 + s.mu.Unlock() + go s.handle(conn, first) + } +} + +// handle drives one connection. On the FIRST connection the mailbox reports one +// message and its BODY[] fetch under-delivers; on any later connection (i.e. +// after a reconnect) the mailbox is empty so the resumed folder completes. +func (s *underflowServer) handle(conn net.Conn, first bool) { + defer func() { _ = conn.Close() }() + br := bufio.NewReader(conn) + fmt.Fprint(conn, "* OK IMAP4rev1 ready\r\n") + for { + line, err := br.ReadString('\n') + if err != nil { + return + } + fields := strings.Fields(line) + if len(fields) == 0 { + continue + } + tag := fields[0] + up := strings.ToUpper(line) + switch { + case strings.Contains(up, "LOGIN"): + fmt.Fprintf(conn, "%s OK LOGIN completed\r\n", tag) + case strings.Contains(up, "EXAMINE"), strings.Contains(up, "SELECT"): + n := 0 + if first { + n = 1 + } + fmt.Fprintf(conn, "* %d EXISTS\r\n", n) + fmt.Fprint(conn, "* OK [UIDVALIDITY 1] ok\r\n") + fmt.Fprintf(conn, "%s OK [READ-ONLY] EXAMINE completed\r\n", tag) + case strings.Contains(up, "BODY["), strings.Contains(up, "BODY.PEEK"): + // Poison: announce 100000 bytes, send 10, then stall until shutdown. + fmt.Fprint(conn, "* 1 FETCH (UID 1 BODY[] {100000}\r\n") + fmt.Fprint(conn, "0123456789") + <-s.stop + return + case strings.Contains(up, "FETCH"): + // Pass-1 metadata fetch for the single message. + fmt.Fprint(conn, "* 1 FETCH (UID 1 RFC822.SIZE 100 FLAGS () "+ + "INTERNALDATE \"01-Jan-2020 00:00:00 +0000\" "+ + "ENVELOPE (\"Wed, 01 Jan 2020 00:00:00 +0000\" \"poison\" NIL NIL NIL NIL NIL NIL NIL \"\"))\r\n") + fmt.Fprintf(conn, "%s OK FETCH completed\r\n", tag) + case strings.Contains(up, "LOGOUT"): + fmt.Fprintf(conn, "* BYE\r\n%s OK LOGOUT completed\r\n", tag) + return + case strings.Contains(up, "CREATE"), strings.Contains(up, "NOOP"): + fmt.Fprintf(conn, "%s OK completed\r\n", tag) + default: + fmt.Fprintf(conn, "%s OK completed\r\n", tag) + } + } +} + +// TestStreamOneTimesOutOnLiteralUnderflow proves a server that announces a +// larger BODY[] literal than it sends makes streamOne return ErrBodyTimeout +// (bounded by bodyIdleTimeout) instead of hanging forever. +func TestStreamOneTimesOutOnLiteralUnderflow(t *testing.T) { + restore := shortenBodyIdle(200 * time.Millisecond) + defer restore() + + srv := newUnderflowServer(t) + defer srv.close() + ctx := context.Background() + + src, err := Connect(ctx, srv.addr()) + if err != nil { + t.Fatalf("connect: %v", err) + } + defer func() { _ = src.Close() }() + if err := src.Login("u", "p").Wait(); err != nil { + t.Fatalf("login: %v", err) + } + + done := make(chan error, 1) + go func() { + done <- streamOne(src, src, "INBOX", imap.UID(1), nil, time.Time{}, nil) + }() + + select { + case err := <-done: + if !errors.Is(err, ErrBodyTimeout) { + t.Fatalf("want ErrBodyTimeout, got %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("streamOne hung on literal underflow instead of timing out") + } +} + +// TestCopyFolderSkipsAndReconnectsOnUnderflow proves CopyFolder does not hang on +// a poisoned message: it marks the message migrated (so future runs skip it), +// invokes ReconnectSrc, and returns with the error counted — the folder is not +// wedged. +func TestCopyFolderSkipsAndReconnectsOnUnderflow(t *testing.T) { + restore := shortenBodyIdle(200 * time.Millisecond) + defer restore() + + srv := newUnderflowServer(t) + defer srv.close() + ctx := context.Background() + + src, err := Connect(ctx, srv.addr()) + if err != nil { + t.Fatalf("connect: %v", err) + } + defer func() { _ = src.Close() }() + if err := src.Login("u", "p").Wait(); err != nil { + t.Fatalf("login: %v", err) + } + + var marked []string + var reconnected bool + deps := CopyDeps{ + IsMigrated: func(string) (bool, error) { return false, nil }, + MarkMigrated: func(_, k string) error { marked = append(marked, k); return nil }, + OnProgress: func(_, _ int) {}, + ReconnectSrc: func() (*Client, error) { + reconnected = true + nc, derr := Connect(ctx, srv.addr()) + if derr != nil { + return nil, derr + } + if lerr := nc.Login("u", "p").Wait(); lerr != nil { + return nil, lerr + } + return nc, nil + }, + } + + done := make(chan CopyResult, 1) + go func() { + r, _ := CopyFolder(ctx, src, src, "INBOX", "INBOX", deps) + done <- r + }() + + select { + case r := <-done: + if r.Errors == 0 { + t.Fatalf("expected the poisoned message counted as an error, got %+v", r) + } + if !reconnected { + t.Fatal("ReconnectSrc was never called") + } + if len(marked) != 1 { + t.Fatalf("poisoned message should be marked migrated once, got %v", marked) + } + case <-time.After(8 * time.Second): + t.Fatal("CopyFolder hung on a poisoned message instead of skipping it") + } +} + +func shortenBodyIdle(d time.Duration) func() { + old := bodyIdleTimeout + bodyIdleTimeout = d + return func() { bodyIdleTimeout = old } +} diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 1d42325..21e6feb 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -333,10 +333,19 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in if err != nil { return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "src", err) } - defer func() { _ = src.Logout().Wait() }() if err := src.Login(a.SrcLogin, string(srcPass)).Wait(); err != nil { + _ = src.Logout().Wait() return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "src", err) } + // srcClient holds the LIVE source connection. A body-read timeout (server + // under-delivering a literal) forces a mid-run reconnect via reconnectSrc, + // which swaps this pointer. The cancel goroutine and the deferred logout + // below both read through it, so they always act on the current connection + // rather than a stale one that was already replaced and logged out. + var srcClient atomic.Pointer[imapx.Client] + srcClient.Store(src) + defer func() { _ = srcClient.Load().Logout().Wait() }() + dst, err := imapx.Connect(actx, dstEP) if err != nil { return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "dst", err) @@ -346,11 +355,32 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "dst", err) } + // reconnectSrc dials and logs in a fresh source client, swaps it in as the + // live connection, and logs the old (desynced) one out. CopyFolder calls it + // to recover after a message body times out: the server left the connection + // mid-literal, so it can't be reused. Bound to actx, so a cancelled account + // fails the dial instead of reconnecting. + reconnectSrc := func() (*imapx.Client, error) { + nc, err := imapx.Connect(actx, srcEP) + if err != nil { + return nil, err + } + if err := nc.Login(a.SrcLogin, string(srcPass)).Wait(); err != nil { + _ = nc.Logout().Wait() + return nil, err + } + if old := srcClient.Swap(nc); old != nil { + _ = old.Logout().Wait() + } + slog.Info("reconnected src after message body timeout", "account", a.ID, "src_login", a.SrcLogin) + return nc, nil + } + // On cancel, close the connections so any in-flight network read (a slow // FETCH/Collect that ctx.Err() checks can't interrupt) unblocks immediately. go func() { <-actx.Done() - _ = src.Close() + _ = srcClient.Load().Close() _ = dst.Close() }() @@ -445,6 +475,10 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in // stall watchdog sees a large-but-live transfer as progress instead of // cancelling it as a wedged connection. OnActivity: touch, + // Recovers from a message body timeout (server under-delivering a + // literal) by swapping in a fresh source connection so the folder can + // resume with the remaining messages. + ReconnectSrc: reconnectSrc, OnProgress: func(c, s int) { touch() now := time.Now() @@ -491,7 +525,7 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in if actx.Err() != nil { break // cancelled — stop scheduling more folders } - res, err := imapx.CopyFolder(actx, src, dst, fp.src, fp.dst, deps) + res, err := imapx.CopyFolder(actx, srcClient.Load(), dst, fp.src, fp.dst, deps) folderErr := int64(0) if err != nil && actx.Err() == nil { slog.Warn("folder copy error", "account", a.ID, "src_login", a.SrcLogin, "folder", fp.src, "err", err)