diff --git a/internal/imapx/copy.go b/internal/imapx/copy.go index 3c6541e..4aae57e 100644 --- a/internal/imapx/copy.go +++ b/internal/imapx/copy.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "io" + "log/slog" "net" "strings" "time" @@ -35,6 +36,12 @@ type CopyDeps struct { // text — so the orchestrator can persist individual errors for the // per-account error modal. Folder-level errors are reported by the caller. OnError func(ref, msg string) + // OnActivity is called repeatedly WHILE a single message body is streamed + // (each FETCH read chunk and each APPEND write chunk). Copying one large + // message can take longer than the orchestrator's stall timeout; without an + // 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() } // CopyResult summarizes the outcome of one CopyFolder run. @@ -180,7 +187,7 @@ func CopyFolder(ctx context.Context, src, dst *imapclient.Client, srcFolder, dst if err := ctx.Err(); err != nil { return res, err } - 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, deps.OnActivity); err != nil { res.Errors++ reportErr(msgRef(q.uid, q.subject), "copy message: "+err.Error()) // A closed/broken connection won't recover: every remaining APPEND @@ -204,11 +211,50 @@ func CopyFolder(ctx context.Context, src, dst *imapclient.Client, srcFolder, dst return res, nil } +// slowMessage marks how long one message's FETCH or APPEND phase may take +// before it is logged as anomalous. Well below the orchestrator's 3-minute +// stall timeout, so a message that trips the watchdog always leaves a log line +// naming the phase (FETCH vs APPEND) and size — turning a silent stall into +// 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() +} + +func (t touchReader) Read(p []byte) (int, error) { + n, err := t.r.Read(p) + if n > 0 && t.on != nil { + t.on() + } + return n, err +} + +// touchWriter wraps the APPEND write stream and pings onActivity on every +// non-empty write, so a long upload keeps the stall watchdog fed byte-by-byte. +type touchWriter struct { + w io.Writer + on func() +} + +func (t touchWriter) Write(p []byte) (int, error) { + n, err := t.w.Write(p) + if n > 0 && t.on != nil { + t.on() + } + return n, err +} + // streamOne FETCHes BODY[] for one message and APPENDs it into dst without // spooling to disk. The body is buffered in RAM only for the duration of -// this single FETCH->APPEND round trip. -func streamOne(src, dst *imapclient.Client, dstFolder string, uid imap.UID, flags []imap.Flag, internalDate time.Time) error { +// 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 { bodySection := &imap.FetchItemBodySection{} + fetchStart := time.Now() fetchCmd := src.Fetch(imap.UIDSetNum(uid), &imap.FetchOptions{ BodySection: []*imap.FetchItemBodySection{bodySection}, }) @@ -225,7 +271,7 @@ func streamOne(src, dst *imapclient.Client, dstFolder string, uid imap.UID, flag break } if d, ok := item.(imapclient.FetchItemDataBodySection); ok { - b, err := io.ReadAll(d.Literal) + b, err := io.ReadAll(touchReader{r: d.Literal, on: onActivity}) if err != nil { return err } @@ -238,14 +284,16 @@ func streamOne(src, dst *imapclient.Client, dstFolder string, uid imap.UID, flag if body == nil { return fmt.Errorf("empty body uid %v", uid) } + fetchDur := time.Since(fetchStart) + appendStart := time.Now() appendCmd := dst.Append(dstFolder, int64(len(body)), &imap.AppendOptions{Flags: keepFlags(flags), Time: internalDate}) // Append acquires go-imap's per-client encoder mutex and holds it until // Close() calls enc.end(). Close() MUST run on every path: if io.Copy // fails mid-write (server stall, idle timeout), returning without Close() // leaks the mutex and the NEXT Append on this client deadlocks forever on // beginCommand. Close() is idempotent and always releases the lock. - _, copyErr := io.Copy(appendCmd, bytes.NewReader(body)) + _, copyErr := io.Copy(touchWriter{w: appendCmd, on: onActivity}, bytes.NewReader(body)) closeErr := appendCmd.Close() if copyErr != nil { return fmt.Errorf("append body uid %v: %w", uid, copyErr) @@ -253,8 +301,19 @@ func streamOne(src, dst *imapclient.Client, dstFolder string, uid imap.UID, flag if closeErr != nil { return closeErr } - _, err := appendCmd.Wait() - return err + if _, err := appendCmd.Wait(); err != nil { + return err + } + appendDur := time.Since(appendStart) + + // One message that individually eats a large slice of the stall budget is + // the prime suspect behind a "no progress" cancel; name it, its size, and + // which phase was slow so the culprit is visible in the logs. + if fetchDur > slowMessage || appendDur > slowMessage { + slog.Warn("slow message copy", "uid", uid, "bytes", len(body), + "fetch", fetchDur.Round(time.Millisecond), "append", appendDur.Round(time.Millisecond)) + } + return nil } // keepFlags drops \Recent: it cannot be set via APPEND. go-imap v2 beta.8 diff --git a/internal/imapx/copy_test.go b/internal/imapx/copy_test.go index 6c73cee..f384774 100644 --- a/internal/imapx/copy_test.go +++ b/internal/imapx/copy_test.go @@ -169,6 +169,56 @@ func TestCopyFolderPreservesInternalDate(t *testing.T) { } } +// TestCopyFolderReportsActivityDuringBody proves CopyFolder invokes OnActivity +// while a message body is being transferred (FETCH/APPEND), not only between +// messages. This is what keeps the orchestrator's stall watchdog from killing a +// single large-but-live message whose transfer legitimately exceeds the stall +// timeout: without an in-body activity signal, one slow message looks identical +// to a wedged connection. +func TestCopyFolderReportsActivityDuringBody(t *testing.T) { + ep := testEP(t) + ctx := context.Background() + + seedInbox(t, ep, "actsrc@localhost", "p", 1) + + src, err := Connect(ctx, ep) + if err != nil { + t.Fatal(err) + } + defer func() { _ = src.Logout().Wait() }() + if err := src.Login("actsrc@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("actdst@localhost", "p").Wait(); err != nil { + t.Fatal(err) + } + + var activity int + deps := CopyDeps{ + IsMigrated: func(string) (bool, error) { return false, nil }, + MarkMigrated: func(_, _ string) error { return nil }, + OnProgress: func(_, _ int) {}, + OnActivity: func() { activity++ }, + } + + r, err := CopyFolder(ctx, src, dst, "INBOX", "INBOX", deps) + if err != nil { + t.Fatalf("CopyFolder: %v", err) + } + if r.Copied != 1 { + t.Fatalf("copied=%d want 1", r.Copied) + } + if activity == 0 { + t.Fatal("OnActivity never called during body transfer") + } +} + // Требует два ящика на greenmail. Первый запуск копирует N, второй — 0 (все skipped). func TestCopyFolderIdempotent(t *testing.T) { ep := testEP(t) // plain greenmail diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 7038112..1d42325 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -306,9 +306,9 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in "dst_login": a.DstLogin, "dst_host": dstEP.Host, "dst_port": dstEP.Port, }}) _ = o.store.SetAccountStatus(ctx, a.ID, "running") - _ = o.store.SetAccountError(ctx, a.ID, "") // clear any error from a previous run - _ = o.store.ResetAccountCounters(ctx, a.ID) // start from zero; IncAccountCounters is additive - _ = o.store.ClearAccountErrors(ctx, a.ID) // drop last run's per-error rows + _ = o.store.SetAccountError(ctx, a.ID, "") // clear any error from a previous run + _ = o.store.ResetAccountCounters(ctx, a.ID) // start from zero; IncAccountCounters is additive + _ = o.store.ClearAccountErrors(ctx, a.ID) // drop last run's per-error rows // Per-account cancellable context: IMAP work uses actx (so CancelAccount // stops it); DB writes keep the parent ctx so status/counters persist even @@ -441,6 +441,10 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in IsMigrated: func(k string) (bool, error) { return o.store.IsMigrated(ctx, a.ID, 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) }, + // Fires as bytes move within a single message's FETCH/APPEND, so the + // stall watchdog sees a large-but-live transfer as progress instead of + // cancelling it as a wedged connection. + OnActivity: touch, OnProgress: func(c, s int) { touch() now := time.Now()