controller/MailMutationPipeline.h - GCC Code Coverage Report

Directory:	src/
Coverage:	low: ≥ 0% medium: ≥ 75.0% high: ≥ 90.0%

	Coverage	Exec / Excl / Total
Lines:	100.0%	11 / 0 / 11
Functions:	100.0%	11 / 0 / 11
Branches:	-%	0 / 0 / 0

    controller/MailMutationPipeline.h
    
        Line
        Branch
        Exec
        Source
      
        #pragma once
      
        #include <QHash>
      
        #include <QList>
      
        #include <QMetaObject>
      
        #include <QObject>
      
        #include <QPointer>
      
        #include <QQueue>
      
        #include <QSet>
      
        #include <QString>
      
        #include <QThreadPool>
      
        #include <functional>
      
        #include <optional>
      
        #include "controller/FolderContext.h"
      
        #include "controller/PendingFlagIntents.h"
      
        #include "data/AccountConfig.h"
      
        #include "data/MailCache.h"
      
        #include "data/Models.h"
      
        #include "data/MutationJournal.h"
      
        class ImapService;
      
        class MailListModel;
      
        class MailThreadModel;
      
        // Durable mutation pipeline of the controller (Sprint 91 S91.4.2, extracted
      
        // from MailController): desired-state journaling, APPEND queueing with the
      
        // bounded payload lanes, journal replay and dispatch reservations (F4.3
      
        // batched moves), completion handling, readback verification and
      
        // ambiguous-outcome reconciliation. The durability contract is
      
        // local intent -> durable -> dispatched -> confirmed/failed. Callers reach it
      
        // as `controller.mutations()`.
      
        //
      
        // Owner: MailController (value member, declared before FolderFlagSync so
      
        // the flag sync can bind to intents()). GUI thread; APPEND payload reads and
      
        // writes run on the unit's two one-thread pools. The main and body
      
        // connections, their lifecycle, the folder context, IMAP config, models and
      
        // the unavailable-folder set stay owned by the controller and are read by
      
        // reference. The controller wires the connections by calling
      
        // connectMainConnection() / connectMutationVerification() /
      
        // connectCacheSignals() at the former positions, so receiver order is
      
        // unchanged. Controller signals keep their emitter (notifiers in Deps).
      
        class MailMutationPipeline final : public QObject {
      
        50
          Q_OBJECT
      
        public:
      
          struct Deps {
      
            std::function<void()> ensureBodyConnection;
      
            // MailController signals (the emitter stays the controller).
      
            std::function<void(const QString &)> statusMessage;
      
            std::function<void(const QString &)> actionError;
      
            std::function<void(qint64 token, const QString &message)> pendingActionChanged;
      
            std::function<void(qint64 token, const QString &folderPath, qint64 uid)>
      
                appendMutationCompleted;
      
            std::function<void(qint64 token, const QString &error)> appendMutationFailed;
      
            std::function<void(const QString &folder, int count)> unreadCountChanged;
      
            std::function<void(const QString &folder, int count)> cachedUnreadCountChanged;
      
          };
      
          MailMutationPipeline(ImapService *imap, ImapService *const &bodyImap, MailCache *cache,
      
                               MailListModel *model, MailThreadModel *const &threadModel,
      
                               const FolderContext &folder, const ImapConfig &imapConfig,
      
                               const QSet<QString> &unavailableFolders, Deps deps,
      
                               QObject *parent = nullptr);
      
          // Wiring, called by the controller at the former connection positions.
      
          void connectMainConnection(ImapService *imap);
      
          void connectMutationVerification(ImapService *bodyImap);
      
          void connectCacheSignals();
      
          // Account switch: drop verification/ambiguity state of the old account.
      
          void resetForAccountSwitch();
      
          // ~MailController: cancel payload reads and join both APPEND lanes.
      
          void shutdown();
      
          // S91.0.7: pending flag/move intents, readback fences and last polled
      
          // unread counts — one owner, accessed through query/mark methods.
      
        3908
          PendingFlagIntents &intents() { return m_flagIntents; }
      
        84
          const PendingFlagIntents &intents() const { return m_flagIntents; }
      
          qint64 persistDesiredMutation(qint64 folderId, qint64 uid,
      
                                        MutationKind kind, const QString &flag,
      
                                        bool desiredState);
      
          bool retryAppendMutation(qint64 token);
      
          qint64 queueAppend(const QString &folderPath,
      
                             const QByteArray &rfcMessage, const QString &flags,
      
                             const QString &messageId,
      
                             const QByteArray &precomputedSha256 = {});
      
          // Accepted bytes are durably journaled on a bounded writer before the
      
          // callback receives their token. A failed write reports token 0.
      
          void queueAppendAsync(const QString &folderPath, const QByteArray &rfcMessage,
      
                                const QString &flags, const QString &messageId,
      
                                const QByteArray &precomputedSha256,
      
                                std::function<void(qint64)> callback);
      
          void dispatchMutation(const MutationRecord &record);
      
          void replayMutationJournal();
      
          void cancelPendingAppendPayloadReads(ImapService *service = nullptr);
      
          // F4.3: batched moves stay well under the conservative 8192-octet IMAP
      
          // command-line floor (100 uids ≈ 1.3 kB with margins for paths/quoting).
      
          static constexpr int MaxBatchedMoveMessages = 100;
      
          // Verification/reconciliation state contract (tests drive the pipeline
      
          // without a live connection; S91.4.2 replaced friend access, rule 8).
      
        2529
          QSet<qint64> &scheduledForDispatch() { return m_mutationsScheduledForDispatch; }
      
        29
          QSet<qint64> &scheduledForVerification() { return m_mutationsScheduledForVerification; }
      
        2
          const QSet<qint64> &deferredVerifications() const { return m_deferredMutationVerifications; }
      
        3
          const QSet<qint64> &resentMutations() const { return m_resentMutations; }
      
        9
          QQueue<qint64> &ambiguityQueue() { return m_ambiguousReconciliationQueue; }
      
        10
          qint64 activeReconciliationToken() const { return m_activeReconciliationToken; }
      
        2
          void setActiveReconciliationToken(qint64 token) { m_activeReconciliationToken = token; }
      
          // How long a dispatch or readback SELECT fence waits for its folder before
      
          // the watchdog hands the reservation back (15 s; tests shorten it).
      
        1
          void setSelectFenceTimeoutMs(int ms) { m_selectFenceTimeoutMs = ms; }
      
          void reportSlowVerification(const MutationRecord &record);
      
          void reconcileNextAmbiguousMutation();
      
          void reconcileAmbiguousMutation(const MutationRecord &record);
      
          void finishAmbiguousReconciliation();
      
          void confirmReconciliationInTarget(const MutationRecord &record,
      
                                             const QList<qint64> &targetUids);
      
          void checkAppendPayloadForReconciliation(const MutationRecord &record);
      
          void searchReconciliationSource(const MutationRecord &record);
      
        public slots:
      
          void onMutationDispatched(qint64 token);
      
          void onMutationCompleted(qint64 token, int outcome,
      
                                   const QString &error,
      
                                   const QString &uidPlusEvidence);
      
          void onFlagSnapshotsReceived(const QList<ImapFlagSnapshot> &snapshots,
      
                                       const QString &folderPath);
      
          void onVerificationFlagsReceived(const QList<ImapFlagSnapshot> &snapshots,
      
                                           const QString &folderPath, qint64 requestedUid);
      
        private:
      
          // onlyUid != 0: a one-UID verification readback (S106-195).
      
          void applyFlagSnapshots(ImapService *origin, const QList<ImapFlagSnapshot> &snapshots,
      
                                  const QString &folderPath, qint64 onlyUid);
      
          bool admitMutationForDispatch(const MutationRecord &record);
      
          static bool selectedIdentityMatches(const ImapService *service, const MutationRecord &record);
      
          void dispatchMoveBatch(const QList<MutationRecord> &batch);
      
          void startAppendPayloadRead(ImapService *service, qint64 token,
      
                                      const MutationRecord &record);
      
          // F4.3: collect the maximal same-source/target/epoch journal move batch
      
          // behind `record` for one shared UID MOVE uid-set command.
      
          QList<MutationRecord> batchableMoveRecords(const MutationRecord &record) const;
      
          // F4.3: release a dispatch reservation that will not reach the wire on
      
          // its own (claim failure, stalled folder select) and re-drive the journal.
      
          void releaseDispatchReservation(qint64 token);
      
          void reserveAndSelectForDispatch(ImapService *service,
      
                                           const QList<qint64> &tokens,
      
                                           const QString &folderPath,
      
                                           const std::function<void()> &dispatch);
      
          void deferMutationVerification(const MutationRecord &record);
      
          void retireSourceFlagVerification(const MutationRecord &record);
      
          void requestMutationVerification(const MutationRecord &record);
      
          void rollbackDesiredMutation(const MutationRecord &record);
      
          ImapService *m_imap;
      
          ImapService *const &m_bodyImap;          // MailController (lazy body connection)
      
          MailCache *m_cache;
      
          MailListModel *m_model;
      
          MailThreadModel *const &m_threadModel;   // MailController::setThreadModel
      
          const FolderContext &m_folder;           // MailController::m_folder
      
          const ImapConfig &m_imapConfig;          // MailController::setImapConfig
      
          const QSet<QString> &m_unavailableFolders;
      
          Deps m_deps;
      
          int m_selectFenceTimeoutMs = 15000;
      
          PendingFlagIntents m_flagIntents;
      
          QSet<qint64> m_mutationsScheduledForVerification;
      
          QSet<qint64> m_deferredMutationVerifications;
      
          QHash<qint64, int> m_mutationVerificationAttempts;
      
          // Desired-state tokens already re-sent once after a differing readback.
      
          QSet<qint64> m_resentMutations;
      
          QSet<qint64> m_slowVerificationReports;
      
          quint64 m_mutationVerificationGeneration = 0;
      
          QSet<qint64> m_mutationsScheduledForDispatch;
      
          // Reserved APPENDs whose payload is still being read on the dedicated lane.
      
          // The durable claim waits until immediately before the wire command, so
      
          // cancellation or a crash leaves these tokens safely Queued.
      
          struct PendingAppendPayloadRead {
      
            quint64 sequence = 0;
      
            QPointer<ImapService> service;
      
          };
      
          struct AppendPayloadReadResult {
      
            std::optional<MailCache::AppendPayload> payload;
      
            QString error;
      
          };
      
          struct AppendPayloadCheckResult {
      
            bool valid = false;
      
            QString error;
      
          };
      
          QHash<qint64, PendingAppendPayloadRead> m_appendPayloadReads;
      
          quint64 m_appendPayloadReadSequence = 0;
      
          QThreadPool m_appendPayloadPool;
      
          QThreadPool m_appendWritePool;
      
          qint64 m_pendingAppendWriteBytes = 0;
      
          int m_pendingAppendWrites = 0;
      
          QQueue<qint64> m_ambiguousReconciliationQueue;
      
          qint64 m_activeReconciliationToken = 0;
      
          quint64 m_reconciliationSequence = 0;
      
          QList<QMetaObject::Connection> m_reconciliationConnections;
      
        };

Line	Exec	Source
1		#pragma once
2
3		#include <QHash>
4		#include <QList>
5		#include <QMetaObject>
6		#include <QObject>
7		#include <QPointer>
8		#include <QQueue>
9		#include <QSet>
10		#include <QString>
11		#include <QThreadPool>
12		#include <functional>
13		#include <optional>
14
15		#include "controller/FolderContext.h"
16		#include "controller/PendingFlagIntents.h"
17		#include "data/AccountConfig.h"
18		#include "data/MailCache.h"
19		#include "data/Models.h"
20		#include "data/MutationJournal.h"
21
22		class ImapService;
23		class MailListModel;
24		class MailThreadModel;
25
26		// Durable mutation pipeline of the controller (Sprint 91 S91.4.2, extracted
27		// from MailController): desired-state journaling, APPEND queueing with the
28		// bounded payload lanes, journal replay and dispatch reservations (F4.3
29		// batched moves), completion handling, readback verification and
30		// ambiguous-outcome reconciliation. The durability contract is
31		// local intent -> durable -> dispatched -> confirmed/failed. Callers reach it
32		// as `controller.mutations()`.
33		//
34		// Owner: MailController (value member, declared before FolderFlagSync so
35		// the flag sync can bind to intents()). GUI thread; APPEND payload reads and
36		// writes run on the unit's two one-thread pools. The main and body
37		// connections, their lifecycle, the folder context, IMAP config, models and
38		// the unavailable-folder set stay owned by the controller and are read by
39		// reference. The controller wires the connections by calling
40		// connectMainConnection() / connectMutationVerification() /
41		// connectCacheSignals() at the former positions, so receiver order is
42		// unchanged. Controller signals keep their emitter (notifiers in Deps).
43		class MailMutationPipeline final : public QObject {
44	50	Q_OBJECT
45		public:
46		struct Deps {
47		std::function<void()> ensureBodyConnection;
48		// MailController signals (the emitter stays the controller).
49		std::function<void(const QString &)> statusMessage;
50		std::function<void(const QString &)> actionError;
51		std::function<void(qint64 token, const QString &message)> pendingActionChanged;
52		std::function<void(qint64 token, const QString &folderPath, qint64 uid)>
53		appendMutationCompleted;
54		std::function<void(qint64 token, const QString &error)> appendMutationFailed;
55		std::function<void(const QString &folder, int count)> unreadCountChanged;
56		std::function<void(const QString &folder, int count)> cachedUnreadCountChanged;
57		};
58		MailMutationPipeline(ImapService imap, ImapService const &bodyImap, MailCache *cache,
59		MailListModel model, MailThreadModel const &threadModel,
60		const FolderContext &folder, const ImapConfig &imapConfig,
61		const QSet<QString> &unavailableFolders, Deps deps,
62		QObject *parent = nullptr);
63
64		// Wiring, called by the controller at the former connection positions.
65		void connectMainConnection(ImapService *imap);
66		void connectMutationVerification(ImapService *bodyImap);
67		void connectCacheSignals();
68		// Account switch: drop verification/ambiguity state of the old account.
69		void resetForAccountSwitch();
70		// ~MailController: cancel payload reads and join both APPEND lanes.
71		void shutdown();
72
73		// S91.0.7: pending flag/move intents, readback fences and last polled
74		// unread counts — one owner, accessed through query/mark methods.
75	3908	PendingFlagIntents &intents() { return m_flagIntents; }
76	84	const PendingFlagIntents &intents() const { return m_flagIntents; }
77
78		qint64 persistDesiredMutation(qint64 folderId, qint64 uid,
79		MutationKind kind, const QString &flag,
80		bool desiredState);
81		bool retryAppendMutation(qint64 token);
82		qint64 queueAppend(const QString &folderPath,
83		const QByteArray &rfcMessage, const QString &flags,
84		const QString &messageId,
85		const QByteArray &precomputedSha256 = {});
86		// Accepted bytes are durably journaled on a bounded writer before the
87		// callback receives their token. A failed write reports token 0.
88		void queueAppendAsync(const QString &folderPath, const QByteArray &rfcMessage,
89		const QString &flags, const QString &messageId,
90		const QByteArray &precomputedSha256,
91		std::function<void(qint64)> callback);
92		void dispatchMutation(const MutationRecord &record);
93		void replayMutationJournal();
94		void cancelPendingAppendPayloadReads(ImapService *service = nullptr);
95		// F4.3: batched moves stay well under the conservative 8192-octet IMAP
96		// command-line floor (100 uids ≈ 1.3 kB with margins for paths/quoting).
97		static constexpr int MaxBatchedMoveMessages = 100;
98
99		// Verification/reconciliation state contract (tests drive the pipeline
100		// without a live connection; S91.4.2 replaced friend access, rule 8).
101	2529	QSet<qint64> &scheduledForDispatch() { return m_mutationsScheduledForDispatch; }
102	29	QSet<qint64> &scheduledForVerification() { return m_mutationsScheduledForVerification; }
103	2	const QSet<qint64> &deferredVerifications() const { return m_deferredMutationVerifications; }
104	3	const QSet<qint64> &resentMutations() const { return m_resentMutations; }
105	9	QQueue<qint64> &ambiguityQueue() { return m_ambiguousReconciliationQueue; }
106	10	qint64 activeReconciliationToken() const { return m_activeReconciliationToken; }
107	2	void setActiveReconciliationToken(qint64 token) { m_activeReconciliationToken = token; }
108		// How long a dispatch or readback SELECT fence waits for its folder before
109		// the watchdog hands the reservation back (15 s; tests shorten it).
110	1	void setSelectFenceTimeoutMs(int ms) { m_selectFenceTimeoutMs = ms; }
111		void reportSlowVerification(const MutationRecord &record);
112		void reconcileNextAmbiguousMutation();
113		void reconcileAmbiguousMutation(const MutationRecord &record);
114		void finishAmbiguousReconciliation();
115		void confirmReconciliationInTarget(const MutationRecord &record,
116		const QList<qint64> &targetUids);
117		void checkAppendPayloadForReconciliation(const MutationRecord &record);
118		void searchReconciliationSource(const MutationRecord &record);
119
120		public slots:
121		void onMutationDispatched(qint64 token);
122		void onMutationCompleted(qint64 token, int outcome,
123		const QString &error,
124		const QString &uidPlusEvidence);
125		void onFlagSnapshotsReceived(const QList<ImapFlagSnapshot> &snapshots,
126		const QString &folderPath);
127		void onVerificationFlagsReceived(const QList<ImapFlagSnapshot> &snapshots,
128		const QString &folderPath, qint64 requestedUid);
129
130		private:
131		// onlyUid != 0: a one-UID verification readback (S106-195).
132		void applyFlagSnapshots(ImapService *origin, const QList<ImapFlagSnapshot> &snapshots,
133		const QString &folderPath, qint64 onlyUid);
134		bool admitMutationForDispatch(const MutationRecord &record);
135		static bool selectedIdentityMatches(const ImapService *service, const MutationRecord &record);
136		void dispatchMoveBatch(const QList<MutationRecord> &batch);
137		void startAppendPayloadRead(ImapService *service, qint64 token,
138		const MutationRecord &record);
139		// F4.3: collect the maximal same-source/target/epoch journal move batch
140		// behind `record` for one shared UID MOVE uid-set command.
141		QList<MutationRecord> batchableMoveRecords(const MutationRecord &record) const;
142		// F4.3: release a dispatch reservation that will not reach the wire on
143		// its own (claim failure, stalled folder select) and re-drive the journal.
144		void releaseDispatchReservation(qint64 token);
145		void reserveAndSelectForDispatch(ImapService *service,
146		const QList<qint64> &tokens,
147		const QString &folderPath,
148		const std::function<void()> &dispatch);
149		void deferMutationVerification(const MutationRecord &record);
150		void retireSourceFlagVerification(const MutationRecord &record);
151		void requestMutationVerification(const MutationRecord &record);
152		void rollbackDesiredMutation(const MutationRecord &record);
153
154		ImapService *m_imap;
155		ImapService *const &m_bodyImap; // MailController (lazy body connection)
156		MailCache *m_cache;
157		MailListModel *m_model;
158		MailThreadModel *const &m_threadModel; // MailController::setThreadModel
159		const FolderContext &m_folder; // MailController::m_folder
160		const ImapConfig &m_imapConfig; // MailController::setImapConfig
161		const QSet<QString> &m_unavailableFolders;
162		Deps m_deps;
163
164		int m_selectFenceTimeoutMs = 15000;
165		PendingFlagIntents m_flagIntents;
166		QSet<qint64> m_mutationsScheduledForVerification;
167		QSet<qint64> m_deferredMutationVerifications;
168		QHash<qint64, int> m_mutationVerificationAttempts;
169		// Desired-state tokens already re-sent once after a differing readback.
170		QSet<qint64> m_resentMutations;
171		QSet<qint64> m_slowVerificationReports;
172		quint64 m_mutationVerificationGeneration = 0;
173		QSet<qint64> m_mutationsScheduledForDispatch;
174		// Reserved APPENDs whose payload is still being read on the dedicated lane.
175		// The durable claim waits until immediately before the wire command, so
176		// cancellation or a crash leaves these tokens safely Queued.
177		struct PendingAppendPayloadRead {
178		quint64 sequence = 0;
179		QPointer<ImapService> service;
180		};
181		struct AppendPayloadReadResult {
182		std::optional<MailCache::AppendPayload> payload;
183		QString error;
184		};
185		struct AppendPayloadCheckResult {
186		bool valid = false;
187		QString error;
188		};
189		QHash<qint64, PendingAppendPayloadRead> m_appendPayloadReads;
190		quint64 m_appendPayloadReadSequence = 0;
191		QThreadPool m_appendPayloadPool;
192		QThreadPool m_appendWritePool;
193		qint64 m_pendingAppendWriteBytes = 0;
194		int m_pendingAppendWrites = 0;
195		QQueue<qint64> m_ambiguousReconciliationQueue;
196		qint64 m_activeReconciliationToken = 0;
197		quint64 m_reconciliationSequence = 0;
198		QList<QMetaObject::Connection> m_reconciliationConnections;
199		};
200