fix(send_queue): 修复重复消息 ID 的失败重排并清理 MIME
- 投递成功后清空 `mime_base64`,避免已投递队列项保留原始 MIME。 - 当同一 `message_id` 的旧队列项已达到最大失败次数时,允许复用并重新入队。 - 补充测试覆盖已投递项清理与终态失败重排场景。
This commit is contained in:
@@ -894,6 +894,13 @@ func TestSendQueueRecoversStaleSendingItems(t *testing.T) {
|
|||||||
if status != sendQueueStatusDelivered {
|
if status != sendQueueStatusDelivered {
|
||||||
t.Fatalf("queue status=%q, want delivered", status)
|
t.Fatalf("queue status=%q, want delivered", status)
|
||||||
}
|
}
|
||||||
|
var mimeBase64 string
|
||||||
|
if err := a.db.QueryRow(`SELECT mime_base64 FROM send_queue WHERE id=?`, queueID).Scan(&mimeBase64); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if mimeBase64 != "" {
|
||||||
|
t.Fatal("delivered queue item should not retain raw MIME")
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestSubmissionAuthRequiresMailboxPasswordAndSendPermission(t *testing.T) {
|
func TestSubmissionAuthRequiresMailboxPasswordAndSendPermission(t *testing.T) {
|
||||||
@@ -1082,6 +1089,46 @@ func TestSubmissionSentCopyDedupesByMessageID(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestSubmissionRequeuesTerminalFailedDuplicateMessageID(t *testing.T) {
|
||||||
|
a := newTestApp(t)
|
||||||
|
a.cfg.SMTPHost = "127.0.0.1"
|
||||||
|
a.cfg.SMTPPort = "1"
|
||||||
|
user, mb, err := a.authenticateSubmission(context.Background(), "admin@lanqin.local", "ChangeMe123!")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
raw := "From: admin@lanqin.local\r\nTo: person@example.com\r\nSubject: requeue\r\nMessage-ID: <requeue@example.test>\r\n\r\nbody"
|
||||||
|
if err := a.submitSMTPMessage(context.Background(), user, mb, mb.Address, []string{"person@example.com"}, strings.NewReader(raw)); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if _, err := a.db.Exec(`UPDATE send_queue SET status=?,attempt_count=max_attempts,next_attempt_at=?,last_error='terminal' WHERE mailbox_id=? AND message_id=?`, sendQueueStatusFailed, a.now().UTC().Add(time.Hour).Format(time.RFC3339Nano), mb.ID, "<requeue@example.test>"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
host, port, received := startCapturingSMTP(t, 1)
|
||||||
|
a.cfg.SMTPHost = host
|
||||||
|
a.cfg.SMTPPort = port
|
||||||
|
if err := a.submitSMTPMessage(context.Background(), user, mb, mb.Address, []string{"person@example.com"}, strings.NewReader(raw)); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := a.processDueSendQueue(context.Background()); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-received:
|
||||||
|
case <-time.After(2 * time.Second):
|
||||||
|
t.Fatal("requeued terminal failure was not relayed")
|
||||||
|
}
|
||||||
|
var status string
|
||||||
|
var attemptCount int
|
||||||
|
if err := a.db.QueryRow(`SELECT status,attempt_count FROM send_queue WHERE mailbox_id=? AND message_id=?`, mb.ID, "<requeue@example.test>").Scan(&status, &attemptCount); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if status != sendQueueStatusDelivered || attemptCount != 1 {
|
||||||
|
t.Fatalf("queue status=%q attempts=%d, want delivered attempts=1", status, attemptCount)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestSubmissionAllowsAuthorizedAliasSendAs(t *testing.T) {
|
func TestSubmissionAllowsAuthorizedAliasSendAs(t *testing.T) {
|
||||||
a := newTestApp(t)
|
a := newTestApp(t)
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
|
|||||||
@@ -66,18 +66,38 @@ func (a *App) enqueueSend(ctx context.Context, in sendQueueInput) (string, error
|
|||||||
}
|
}
|
||||||
id := newID("snd")
|
id := newID("snd")
|
||||||
messageID := strings.TrimSpace(in.MessageID)
|
messageID := strings.TrimSpace(in.MessageID)
|
||||||
|
mimeBase64 := base64.StdEncoding.EncodeToString(in.MIMEBytes)
|
||||||
|
recipientsJSON := jsonEncode(dedupeEmails(in.Recipients))
|
||||||
_, err := a.db.ExecContext(ctx, `INSERT OR IGNORE INTO send_queue(id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,status,next_attempt_at,created_at,updated_at)
|
_, err := a.db.ExecContext(ctx, `INSERT OR IGNORE INTO send_queue(id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,status,next_attempt_at,created_at,updated_at)
|
||||||
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
|
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
|
||||||
id, in.UserID, in.MailboxID, in.SentMessageID, messageID, in.Source, normalizeEmail(in.MailFrom), normalizeEmail(in.HeaderFrom), jsonEncode(dedupeEmails(in.Recipients)), base64.StdEncoding.EncodeToString(in.MIMEBytes), sendQueueStatusQueued, now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano))
|
id, in.UserID, in.MailboxID, in.SentMessageID, messageID, in.Source, normalizeEmail(in.MailFrom), normalizeEmail(in.HeaderFrom), recipientsJSON, mimeBase64, sendQueueStatusQueued, now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
if strings.TrimSpace(messageID) != "" {
|
if strings.TrimSpace(messageID) != "" {
|
||||||
var existingID string
|
var existingID, status string
|
||||||
if err := a.db.QueryRowContext(ctx, `SELECT id FROM send_queue WHERE mailbox_id=? AND source=? AND message_id=?`, in.MailboxID, in.Source, messageID).Scan(&existingID); err != nil {
|
var attemptCount, maxAttempts int
|
||||||
|
if err := a.db.QueryRowContext(ctx, `SELECT id,status,attempt_count,max_attempts FROM send_queue WHERE mailbox_id=? AND source=? AND message_id=?`, in.MailboxID, in.Source, messageID).Scan(&existingID, &status, &attemptCount, &maxAttempts); err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
if existingID != id {
|
if existingID != id {
|
||||||
|
if status == sendQueueStatusFailed && attemptCount >= maxAttempts {
|
||||||
|
_, err := a.db.ExecContext(ctx, `UPDATE send_queue SET user_id=?,sent_message_id=?,mail_from=?,header_from=?,recipients_json=?,mime_base64=?,status=?,attempt_count=0,next_attempt_at=?,last_error='',updated_at=?,delivered_at=NULL WHERE id=? AND status=? AND attempt_count>=max_attempts`,
|
||||||
|
in.UserID, in.SentMessageID, normalizeEmail(in.MailFrom), normalizeEmail(in.HeaderFrom), recipientsJSON, mimeBase64, sendQueueStatusQueued, now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), existingID, sendQueueStatusFailed)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
a.recordSendAudit(ctx, sendAuditQueued, sendQueueStatusQueued, sendAuditInput{
|
||||||
|
QueueID: existingID,
|
||||||
|
UserID: in.UserID,
|
||||||
|
MailboxID: in.MailboxID,
|
||||||
|
SentMessageID: in.SentMessageID,
|
||||||
|
Source: in.Source,
|
||||||
|
MailFrom: in.MailFrom,
|
||||||
|
HeaderFrom: in.HeaderFrom,
|
||||||
|
Recipients: in.Recipients,
|
||||||
|
})
|
||||||
|
}
|
||||||
return existingID, nil
|
return existingID, nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -191,7 +211,7 @@ func (a *App) processSendQueueItem(ctx context.Context, id string) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
now := a.now().UTC().Format(time.RFC3339Nano)
|
now := a.now().UTC().Format(time.RFC3339Nano)
|
||||||
if _, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,delivered_at=?,updated_at=?,last_error='' WHERE id=?`, sendQueueStatusDelivered, now, now, item.ID); err != nil {
|
if _, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,delivered_at=?,updated_at=?,last_error='',mime_base64='' WHERE id=?`, sendQueueStatusDelivered, now, now, item.ID); err != nil {
|
||||||
a.log.Warn("failed to mark send queue delivered", "id", item.ID, "error", err)
|
a.log.Warn("failed to mark send queue delivered", "id", item.ID, "error", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user