diff --git a/internal/imapx/conn.go b/internal/imapx/conn.go deleted file mode 100644 index 08f71a0..0000000 --- a/internal/imapx/conn.go +++ /dev/null @@ -1,30 +0,0 @@ -package imapx - -import ( - "net" - "time" -) - -// idleReadTimeout bounds how long a connection may go WITHOUT receiving any -// bytes from the server before the read is aborted. It is an *idle* timeout, -// not a total deadline: every successful read pushes it forward, so a slow but -// 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 = 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 -// the actual blocking network read regardless of encryption. -type idleConn struct { - net.Conn - timeout time.Duration -} - -func (c *idleConn) Read(b []byte) (int, error) { - if c.timeout > 0 { - // Ignore the error: a closed conn will surface it from Read below. - _ = c.Conn.SetReadDeadline(time.Now().Add(c.timeout)) - } - return c.Conn.Read(b) -} diff --git a/internal/imapx/conn_test.go b/internal/imapx/conn_test.go deleted file mode 100644 index 528525a..0000000 --- a/internal/imapx/conn_test.go +++ /dev/null @@ -1,60 +0,0 @@ -package imapx - -import ( - "io" - "net" - "testing" - "time" - - "context" -) - -// A server that sends the IMAP greeting, accepts one command, then goes silent -// forever must NOT wedge the client: the idle read-timeout has to abort the -// blocked read so the command returns an error instead of hanging. This is the -// exact production failure (account stuck in "running" with zero progress). -func TestConnectIdleReadTimeoutUnwedgesSilentServer(t *testing.T) { - ln, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatalf("listen: %v", err) - } - defer ln.Close() - - go func() { - c, err := ln.Accept() - if err != nil { - return - } - defer c.Close() - _, _ = io.WriteString(c, "* OK IMAP4rev2 ready\r\n") - // Read the LOGIN command bytes, then never answer. - buf := make([]byte, 512) - _, _ = c.Read(buf) - // Block until the client gives up and closes the connection. - _, _ = c.Read(buf) - }() - - old := idleReadTimeout - idleReadTimeout = 150 * time.Millisecond - defer func() { idleReadTimeout = old }() - - port := ln.Addr().(*net.TCPAddr).Port - ep := Endpoint{Host: "127.0.0.1", Port: port, TLSMode: "plain"} - - c, err := Connect(context.Background(), ep) - if err != nil { - t.Fatalf("Connect: %v", err) - } - - done := make(chan error, 1) - go func() { done <- c.Login("user", "pass").Wait() }() - - select { - case err := <-done: - if err == nil { - t.Fatal("expected an error from Login against a silent server, got nil") - } - case <-time.After(3 * time.Second): - t.Fatal("Login did not return: idle read-timeout was not enforced (connection wedged)") - } -} diff --git a/internal/imapx/copy.go b/internal/imapx/copy.go index 9b0eeb3..3c6541e 100644 --- a/internal/imapx/copy.go +++ b/internal/imapx/copy.go @@ -3,8 +3,11 @@ package imapx import ( "bytes" "context" + "errors" "fmt" "io" + "net" + "strings" "time" "github.com/emersion/go-imap/v2" @@ -180,6 +183,12 @@ func CopyFolder(ctx context.Context, src, dst *imapclient.Client, srcFolder, dst if err := streamOne(src, dst, dstFolder, q.uid, q.flags, q.internalDate); err != nil { res.Errors++ reportErr(msgRef(q.uid, q.subject), "copy message: "+err.Error()) + // 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. + if errors.Is(err, net.ErrClosed) || strings.Contains(err.Error(), "use of closed network connection") { + return res, fmt.Errorf("dst connection lost in %q: %w", dstFolder, err) + } continue } if err := deps.MarkMigrated(dstFolder, q.key); err != nil { diff --git a/internal/imapx/dial.go b/internal/imapx/dial.go index 459d973..e0c4c86 100644 --- a/internal/imapx/dial.go +++ b/internal/imapx/dial.go @@ -18,51 +18,49 @@ type Endpoint struct { func (e Endpoint) addr() string { return fmt.Sprintf("%s:%d", e.Host, e.Port) } -// dialTimeout bounds establishing the TCP connection (matches go-imap's own -// default). The subsequent idleReadTimeout governs reads once connected. +// dialTimeout bounds establishing the TCP connection. const dialTimeout = 30 * time.Second -// dialOnce establishes one connection and returns a ready *Client whose reads -// are guarded by idleReadTimeout. Unlike imapclient.Dial*, the underlying TCP -// conn is wrapped in idleConn so a server that stops responding mid-command -// unblocks the read instead of hanging forever. ctx bounds the TCP dial. +// dialOnce establishes one connection and returns a ready *Client. ctx bounds +// the TCP dial. We deliberately do NOT impose a socket-level read deadline: +// a blanket read deadline can't tell an idle connection (e.g. dst sitting idle +// 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) { d := &net.Dialer{Timeout: dialTimeout} raw, err := d.DialContext(ctx, "tcp", ep.addr()) if err != nil { return nil, err } - conn := &idleConn{Conn: raw, timeout: idleReadTimeout} switch ep.TLSMode { case "ssl": // NextProtos mirrors imapclient.DialTLS's ALPN advertisement. - tlsConn := tls.Client(conn, &tls.Config{ServerName: ep.Host, NextProtos: []string{"imap"}}) + tlsConn := tls.Client(raw, &tls.Config{ServerName: ep.Host, NextProtos: []string{"imap"}}) if err := tlsConn.HandshakeContext(ctx); err != nil { - _ = conn.Close() + _ = raw.Close() return nil, err } - c := imapclient.New(tlsConn, nil) - return waitGreeting(c) + return waitGreeting(imapclient.New(tlsConn, nil)) case "starttls": opts := &imapclient.Options{TLSConfig: &tls.Config{ServerName: ep.Host}} - c, err := imapclient.NewStartTLS(conn, opts) + c, err := imapclient.NewStartTLS(raw, opts) if err != nil { return nil, err } return c, nil case "plain": - c := imapclient.New(conn, nil) - return waitGreeting(c) + return waitGreeting(imapclient.New(raw, nil)) default: - _ = conn.Close() + _ = raw.Close() return nil, fmt.Errorf("unknown tls_mode %q", ep.TLSMode) } } // waitGreeting blocks for the server's initial greeting so a mute server is -// caught at connect time (bounded by idleReadTimeout) rather than at the first -// command. NewStartTLS already awaits the greeting during its STARTTLS upgrade. +// caught at connect time rather than at the first command. NewStartTLS already +// awaits the greeting during its STARTTLS upgrade. func waitGreeting(c *imapclient.Client) (*imapclient.Client, error) { if err := c.WaitGreeting(); err != nil { _ = c.Close() diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 4faec64..eb461cb 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -5,6 +5,7 @@ import ( "errors" "log/slog" "sync" + "sync/atomic" "time" "github.com/vasyansk/imap-copier/internal/crypto" @@ -22,6 +23,16 @@ var ErrAlreadyRunning = errors.New("task already running") // suppressed" note. const maxAccountErrors = 500 +// A running account that emits no scan/copy progress for stallTimeout is wedged +// (silent server mid-FETCH, stalled APPEND). The watchdog cancels it so the +// connections close and the worker unwinds instead of hanging forever. This +// replaces socket-level read deadlines, which can't tell an idle connection +// from a stuck one. The threshold is generous so slow-but-live runs aren't cut. +const ( + stallTimeout = 3 * time.Minute + stallCheckInterval = 30 * time.Second +) + // folderPlan is one source folder scheduled for copy and its destination name. type folderPlan struct { src, dst string @@ -316,7 +327,34 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in _ = dst.Close() }() + // 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 + // server). touch() is called on every progress signal below. + var lastActivity atomic.Int64 + lastActivity.Store(time.Now().UnixNano()) + touch := func() { lastActivity.Store(time.Now().UnixNano()) } + go func() { + t := time.NewTicker(stallCheckInterval) + defer t.Stop() + for { + select { + case <-actx.Done(): + return + case <-t.C: + if time.Since(time.Unix(0, lastActivity.Load())) > stallTimeout { + slog.Warn("account stalled with no progress; cancelling", + "account", a.ID, "src_login", a.SrcLogin, "stall", stallTimeout) + _ = o.store.SetAccountError(ctx, a.ID, "stalled: no progress for "+stallTimeout.String()+", cancelled") + cancel() + return + } + } + } + }() + folders, err := imapx.ListFolders(src) + touch() if err != nil { return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "src", err) } @@ -335,6 +373,7 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in } plan[i].total = n grandTotal += n + touch() } o.hub.Publish(wshub.Event{Type: "plan", TaskID: task.ID, Data: map[string]any{ "account_id": a.ID, "src_login": a.SrcLogin, "folders": len(plan), "total": grandTotal, @@ -367,6 +406,7 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in MarkMigrated: func(folder, k string) error { return o.store.MarkMigrated(ctx, a.ID, folder, k) }, OnError: func(ref, msg string) { addErr("message", curFolder, ref, msg) }, OnProgress: func(c, s int) { + touch() now := time.Now() done := c + s // throttle to ~3/sec per account, but always emit folder completion @@ -386,6 +426,7 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in }, // Fires after EXAMINE (before the long fetch) with the folder's message count. OnFolder: func(srcFolder, dstFolder string, total int64) { + touch() curFolder, curTotal = srcFolder, total o.hub.Publish(wshub.Event{Type: "folder", TaskID: task.ID, Data: map[string]any{ "account_id": a.ID, "src_login": a.SrcLogin, @@ -395,6 +436,7 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in // Fires while streaming metadata (dedup scan) so the UI shows movement // before bodies start copying. Throttled to ~4/sec, always emit the last. OnScan: func(scanned, total int64) { + touch() now := time.Now() if now.Sub(lastScanEmit) < 250*time.Millisecond && scanned < total { return diff --git a/web/src/app.css b/web/src/app.css index 7ba5e4b..7d8d622 100644 --- a/web/src/app.css +++ b/web/src/app.css @@ -327,14 +327,75 @@ white-space: nowrap; } -/* error text inside the per-account errors modal: wrap long messages */ -.err-cell { - max-width: 420px; +/* per-account errors modal: a paginated list so long messages wrap on the + full modal width instead of squeezing into narrow table columns. */ +.err-empty { + padding: 24px 8px; + text-align: center; + color: var(--fg-dim); +} + +.err-list { + list-style: none; + margin: 0; + padding: 0; + max-height: 60vh; + overflow-y: auto; +} + +.err-item { + padding: 10px 4px; + border-bottom: 1px solid var(--border); +} + +.err-item-head { + display: flex; + flex-wrap: wrap; + align-items: baseline; + gap: 8px; + font-size: 11px; + letter-spacing: 0.04em; + margin-bottom: 4px; +} + +.err-kind { + color: var(--accent); + text-transform: uppercase; +} + +.err-folder { + color: var(--fg); +} + +.err-ref { + color: var(--fg-dim); + word-break: break-word; +} + +.err-time { + margin-left: auto; + color: var(--fg-dim); +} + +.err-text { + font-size: 12px; + line-height: 1.45; + color: var(--fail); white-space: normal; word-break: break-word; - color: var(--fail); +} + +.err-pager { + display: flex; + align-items: center; + justify-content: center; + gap: 14px; + padding-top: 12px; +} + +.err-pageinfo { font-size: 12px; - line-height: 1.4; + color: var(--fg-dim); } /* clear-log button: mirrors the .panel-label tab on the right edge */ diff --git a/web/src/components/AccountErrorsModal.tsx b/web/src/components/AccountErrorsModal.tsx index fbb284e..d535141 100644 --- a/web/src/components/AccountErrorsModal.tsx +++ b/web/src/components/AccountErrorsModal.tsx @@ -3,6 +3,7 @@ import { Modal } from './Modal' import { listAccountErrors, type AccountError } from '../api' const fmt = (iso: string) => (iso ? new Date(iso).toLocaleString() : '—') +const PAGE_SIZE = 50 export function AccountErrorsModal({ taskId, @@ -15,51 +16,76 @@ export function AccountErrorsModal({ }) { const [errors, setErrors] = useState(null) const [failed, setFailed] = useState(false) + const [page, setPage] = useState(0) const open = account !== null useEffect(() => { if (!account) return setErrors(null) setFailed(false) + setPage(0) listAccountErrors(taskId, account.id) .then((e) => setErrors(e ?? [])) .catch(() => setFailed(true)) }, [taskId, account]) + const total = errors?.length ?? 0 + const pageCount = Math.max(1, Math.ceil(total / PAGE_SIZE)) + const current = Math.min(page, pageCount - 1) + const slice = errors?.slice(current * PAGE_SIZE, current * PAGE_SIZE + PAGE_SIZE) ?? [] + + const title = account + ? `Errors — ${account.src_login}${total ? ` (${total})` : ''}` + : 'Errors' + return ( - -
- - - - - - - - - - - - {failed ? ( - - ) : errors === null ? ( - - ) : errors.length === 0 ? ( - - ) : ( - errors.map((e) => ( - - - - - - - - )) - )} - -
TimeKindFolderMessageError
failed to load errors
loading…
no errors recorded
{fmt(e.created_at)}{e.kind}{e.folder || '—'}{e.message_ref || '—'}{e.error}
-
+ + {failed ? ( +
failed to load errors
+ ) : errors === null ? ( +
loading…
+ ) : total === 0 ? ( +
no errors recorded
+ ) : ( + <> +
    + {slice.map((e) => ( +
  • +
    + {e.kind} + {e.folder && {e.folder}} + {e.message_ref && {e.message_ref}} + {fmt(e.created_at)} +
    +
    {e.error}
    +
    • + ))} +
+ {pageCount > 1 && ( +
+ + + page {current + 1} / {pageCount} + + +
+ )} + + )} ) }