Remove idle connection timeout handling

Remove error handling for closed connections

Add progress watchdog to detect stalled accounts

Improve error modal styling and pagination
This commit is contained in:
2026-07-05 14:50:25 +07:00
parent 95ddbf5619
commit e84366eb0c
7 changed files with 191 additions and 145 deletions
-30
View File
@@ -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)
}
-60
View File
@@ -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)")
}
}
+9
View File
@@ -3,8 +3,11 @@ package imapx
import ( import (
"bytes" "bytes"
"context" "context"
"errors"
"fmt" "fmt"
"io" "io"
"net"
"strings"
"time" "time"
"github.com/emersion/go-imap/v2" "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 { if err := streamOne(src, dst, dstFolder, q.uid, q.flags, q.internalDate); err != nil {
res.Errors++ res.Errors++
reportErr(msgRef(q.uid, q.subject), "copy message: "+err.Error()) 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 continue
} }
if err := deps.MarkMigrated(dstFolder, q.key); err != nil { if err := deps.MarkMigrated(dstFolder, q.key); err != nil {
+15 -17
View File
@@ -18,51 +18,49 @@ type Endpoint struct {
func (e Endpoint) addr() string { return fmt.Sprintf("%s:%d", e.Host, e.Port) } 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 // dialTimeout bounds establishing the TCP connection.
// default). The subsequent idleReadTimeout governs reads once connected.
const dialTimeout = 30 * time.Second const dialTimeout = 30 * time.Second
// dialOnce establishes one connection and returns a ready *Client whose reads // dialOnce establishes one connection and returns a ready *Client. ctx bounds
// are guarded by idleReadTimeout. Unlike imapclient.Dial*, the underlying TCP // the TCP dial. We deliberately do NOT impose a socket-level read deadline:
// conn is wrapped in idleConn so a server that stops responding mid-command // a blanket read deadline can't tell an idle connection (e.g. dst sitting idle
// unblocks the read instead of hanging forever. ctx bounds the TCP dial. // 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) (*imapclient.Client, error) {
d := &net.Dialer{Timeout: dialTimeout} d := &net.Dialer{Timeout: dialTimeout}
raw, err := d.DialContext(ctx, "tcp", ep.addr()) raw, err := d.DialContext(ctx, "tcp", ep.addr())
if err != nil { if err != nil {
return nil, err return nil, err
} }
conn := &idleConn{Conn: raw, timeout: idleReadTimeout}
switch ep.TLSMode { switch ep.TLSMode {
case "ssl": case "ssl":
// NextProtos mirrors imapclient.DialTLS's ALPN advertisement. // 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 { if err := tlsConn.HandshakeContext(ctx); err != nil {
_ = conn.Close() _ = raw.Close()
return nil, err return nil, err
} }
c := imapclient.New(tlsConn, nil) return waitGreeting(imapclient.New(tlsConn, nil))
return waitGreeting(c)
case "starttls": case "starttls":
opts := &imapclient.Options{TLSConfig: &tls.Config{ServerName: ep.Host}} opts := &imapclient.Options{TLSConfig: &tls.Config{ServerName: ep.Host}}
c, err := imapclient.NewStartTLS(conn, opts) c, err := imapclient.NewStartTLS(raw, opts)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return c, nil return c, nil
case "plain": case "plain":
c := imapclient.New(conn, nil) return waitGreeting(imapclient.New(raw, nil))
return waitGreeting(c)
default: default:
_ = conn.Close() _ = raw.Close()
return nil, fmt.Errorf("unknown tls_mode %q", ep.TLSMode) return nil, fmt.Errorf("unknown tls_mode %q", ep.TLSMode)
} }
} }
// waitGreeting blocks for the server's initial greeting so a mute server is // 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 // caught at connect time rather than at the first command. NewStartTLS already
// command. NewStartTLS already awaits the greeting during its STARTTLS upgrade. // awaits the greeting during its STARTTLS upgrade.
func waitGreeting(c *imapclient.Client) (*imapclient.Client, error) { func waitGreeting(c *imapclient.Client) (*imapclient.Client, error) {
if err := c.WaitGreeting(); err != nil { if err := c.WaitGreeting(); err != nil {
_ = c.Close() _ = c.Close()
+42
View File
@@ -5,6 +5,7 @@ import (
"errors" "errors"
"log/slog" "log/slog"
"sync" "sync"
"sync/atomic"
"time" "time"
"github.com/vasyansk/imap-copier/internal/crypto" "github.com/vasyansk/imap-copier/internal/crypto"
@@ -22,6 +23,16 @@ var ErrAlreadyRunning = errors.New("task already running")
// suppressed" note. // suppressed" note.
const maxAccountErrors = 500 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. // folderPlan is one source folder scheduled for copy and its destination name.
type folderPlan struct { type folderPlan struct {
src, dst string src, dst string
@@ -316,7 +327,34 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in
_ = dst.Close() _ = 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) folders, err := imapx.ListFolders(src)
touch()
if err != nil { if err != nil {
return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "src", err) 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 plan[i].total = n
grandTotal += n grandTotal += n
touch()
} }
o.hub.Publish(wshub.Event{Type: "plan", TaskID: task.ID, Data: map[string]any{ 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, "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) }, 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) }, OnError: func(ref, msg string) { addErr("message", curFolder, ref, msg) },
OnProgress: func(c, s int) { OnProgress: func(c, s int) {
touch()
now := time.Now() now := time.Now()
done := c + s done := c + s
// throttle to ~3/sec per account, but always emit folder completion // 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. // Fires after EXAMINE (before the long fetch) with the folder's message count.
OnFolder: func(srcFolder, dstFolder string, total int64) { OnFolder: func(srcFolder, dstFolder string, total int64) {
touch()
curFolder, curTotal = srcFolder, total curFolder, curTotal = srcFolder, total
o.hub.Publish(wshub.Event{Type: "folder", TaskID: task.ID, Data: map[string]any{ o.hub.Publish(wshub.Event{Type: "folder", TaskID: task.ID, Data: map[string]any{
"account_id": a.ID, "src_login": a.SrcLogin, "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 // Fires while streaming metadata (dedup scan) so the UI shows movement
// before bodies start copying. Throttled to ~4/sec, always emit the last. // before bodies start copying. Throttled to ~4/sec, always emit the last.
OnScan: func(scanned, total int64) { OnScan: func(scanned, total int64) {
touch()
now := time.Now() now := time.Now()
if now.Sub(lastScanEmit) < 250*time.Millisecond && scanned < total { if now.Sub(lastScanEmit) < 250*time.Millisecond && scanned < total {
return return
+66 -5
View File
@@ -327,14 +327,75 @@
white-space: nowrap; white-space: nowrap;
} }
/* error text inside the per-account errors modal: wrap long messages */ /* per-account errors modal: a paginated list so long messages wrap on the
.err-cell { full modal width instead of squeezing into narrow table columns. */
max-width: 420px; .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; white-space: normal;
word-break: break-word; 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; font-size: 12px;
line-height: 1.4; color: var(--fg-dim);
} }
/* clear-log button: mirrors the .panel-label tab on the right edge */ /* clear-log button: mirrors the .panel-label tab on the right edge */
+59 -33
View File
@@ -3,6 +3,7 @@ import { Modal } from './Modal'
import { listAccountErrors, type AccountError } from '../api' import { listAccountErrors, type AccountError } from '../api'
const fmt = (iso: string) => (iso ? new Date(iso).toLocaleString() : '—') const fmt = (iso: string) => (iso ? new Date(iso).toLocaleString() : '—')
const PAGE_SIZE = 50
export function AccountErrorsModal({ export function AccountErrorsModal({
taskId, taskId,
@@ -15,51 +16,76 @@ export function AccountErrorsModal({
}) { }) {
const [errors, setErrors] = useState<AccountError[] | null>(null) const [errors, setErrors] = useState<AccountError[] | null>(null)
const [failed, setFailed] = useState(false) const [failed, setFailed] = useState(false)
const [page, setPage] = useState(0)
const open = account !== null const open = account !== null
useEffect(() => { useEffect(() => {
if (!account) return if (!account) return
setErrors(null) setErrors(null)
setFailed(false) setFailed(false)
setPage(0)
listAccountErrors(taskId, account.id) listAccountErrors(taskId, account.id)
.then((e) => setErrors(e ?? [])) .then((e) => setErrors(e ?? []))
.catch(() => setFailed(true)) .catch(() => setFailed(true))
}, [taskId, account]) }, [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 ( return (
<Modal open={open} title={account ? `Errors — ${account.src_login}` : 'Errors'} onClose={onClose} size="lg"> <Modal open={open} title={title} onClose={onClose} size="lg">
<div className="tbl-wrap"> {failed ? (
<table className="tbl"> <div className="err-empty">failed to load errors</div>
<thead> ) : errors === null ? (
<tr> <div className="err-empty">loading</div>
<th>Time</th> ) : total === 0 ? (
<th>Kind</th> <div className="err-empty">no errors recorded</div>
<th>Folder</th> ) : (
<th>Message</th> <>
<th>Error</th> <ul className="err-list">
</tr> {slice.map((e) => (
</thead> <li key={e.id} className="err-item">
<tbody> <div className="err-item-head">
{failed ? ( <span className="err-kind">{e.kind}</span>
<tr className="empty-row"><td colSpan={5}>failed to load errors</td></tr> {e.folder && <span className="err-folder">{e.folder}</span>}
) : errors === null ? ( {e.message_ref && <span className="err-ref">{e.message_ref}</span>}
<tr className="empty-row"><td colSpan={5}>loading</td></tr> <span className="err-time">{fmt(e.created_at)}</span>
) : errors.length === 0 ? ( </div>
<tr className="empty-row"><td colSpan={5}>no errors recorded</td></tr> <div className="err-text">{e.error}</div>
) : ( </li>
errors.map((e) => ( ))}
<tr key={e.id}> </ul>
<td>{fmt(e.created_at)}</td> {pageCount > 1 && (
<td>{e.kind}</td> <div className="err-pager">
<td>{e.folder || '—'}</td> <button
<td>{e.message_ref || '—'}</td> type="button"
<td className="err-cell">{e.error}</td> className="btn"
</tr> disabled={current === 0}
)) onClick={() => setPage(current - 1)}
)} >
</tbody> prev
</table> </button>
</div> <span className="err-pageinfo">
page {current + 1} / {pageCount}
</span>
<button
type="button"
className="btn"
disabled={current >= pageCount - 1}
onClick={() => setPage(current + 1)}
>
next
</button>
</div>
)}
</>
)}
</Modal> </Modal>
) )
} }