Add activity tracking to prevent stall timeouts during message transfers
Add OnActivity callback to CopyDeps to prevent stall timeouts during large message transfers Implement touchReader and touchWriter wrappers to call OnActivity during FETCH and APPEND operations Add slow message logging to identify performance bottlenecks Add test case to verify activity reporting during message transfers Clean up orchestrator account reset code formatting
This commit is contained in:
+65
-6
@@ -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,9 +301,20 @@ func streamOne(src, dst *imapclient.Client, dstFolder string, uid imap.UID, flag
|
||||
if closeErr != nil {
|
||||
return closeErr
|
||||
}
|
||||
_, err := appendCmd.Wait()
|
||||
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
|
||||
// no longer defines an imap.FlagRecent constant (RFC 9051 dropped \Recent
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user