diff --git a/.gitignore b/.gitignore index 43f2bdc..3a05824 100644 --- a/.gitignore +++ b/.gitignore @@ -47,6 +47,7 @@ # build build/ build-*/ +out/ _codeql_build_dir/ _codeql_detected_source_root diff --git a/CMakeLists.txt b/CMakeLists.txt index 6cdafba..144ec16 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -434,6 +434,7 @@ set(HEADERS include/peripherals/peripheral_interface.h include/console_emulator.h include/backend.h + include/authorization_snapshot.h include/types.h include/url_utils.h include/hardware/aht10.h @@ -593,6 +594,8 @@ if(ENABLE_TESTING) tests/cache_manager_test.cpp tests/controller_test.cpp tests/message_storage_test.cpp + tests/report_delivery_test.cpp + tests/foreground_backend_test.cpp tests/user_cache_test.cpp tests/backlog_worker_test.cpp tests/bounded_executor_test.cpp diff --git a/README.md b/README.md index 08223ba..00d6da5 100644 --- a/README.md +++ b/README.md @@ -104,6 +104,7 @@ If you use MSVC toolchain instead of Ninja/gcc, change the `-G "Ninja"` to your ## Documentation - [State Machine](docs/state_machine.md) - Complete state machine workflow and transitions +- [Foreground Backend Waits](docs/foreground_backend_waits.md) - Configurable waits, cache continuation, and durable background reporting - [Hardware Integration](docs/hardware.md) - Display and NFC wiring/configuration - [Logging System](docs/logging.md) - Detailed logging configuration and usage - [Installation Guide](docs/installation.md) - Complete installation instructions diff --git a/docs/foreground_backend_waits.md b/docs/foreground_backend_waits.md new file mode 100644 index 0000000..8221597 --- /dev/null +++ b/docs/foreground_backend_waits.md @@ -0,0 +1,131 @@ +# Foreground backend waits and durable reporting + +Set `fuelflux::timing::kForegroundBackendWaitTimeout` in +[`include/timing_config.h`](../include/timing_config.h) and rebuild. The default +is `std::chrono::seconds{5}`; a non-positive value fails compilation. This one +setting controls authorization cache fallback, the slow-connection warning, +and release of both refuel and intake transmission screens on every connection +type. HTTP connection/total deadlines and DNS deadlines remain independent. + +## Authorization + +Each attempt captures saved user and tank values, pending-card protection, and +writable local report-storage availability before starting its independent +backend session. The general user/tank snapshot comes from one cache generation. +The captured values stay unchanged for that attempt even if synchronization or +report delivery finishes meanwhile. + +The display initially shows `Проверка...` / `Ожидайте`. An online result received +before the foreground deadline follows the existing allowance, role, and tank +selection checks. A network failure can use the captured snapshot; a definitive +authorization denial cannot. + +When the foreground deadline expires, usable saved data automatically continues +through the existing cache authorization checks. Otherwise the display shows +`Медленное соединение`, `Ожидайте или`, and the keyboard profile's Cancel prompt +(`Нажмите ОТМЕНА (B)` for legacy, `Нажмите ОТМЕНА` for VID). The warning uses small +text lines so it fits the ST7565 as well as the larger ILI9488. Cancel immediately +returns to the welcome screen. Cancel before the warning retains its former +behavior. + +Attempts have independent IDs and sessions. Abandonment signals transport +cancellation, including cancellable c-ares DNS waits on the GSM build. The +controller accepts an attempt's result only while that attempt is current. +A late successful unused session is closed asynchronously; it cannot update +the active session or cache. + +## Reports and subsequent operations + +Before enabling another operation, the controller stores the completed payload +and a protected per-card user/tank snapshot in one SQLite transaction, deducting +the refuel volume exactly once. Intake preserves the allowance. A persistence +failure shows `Ошибка записи` and enters the error procedure instead of showing +successful completion with an unrecorded transaction. + +The transmission screen remains until delivery resolves or the foreground wait +expires. The existing completion screen retains the final volume and accepts a +new card or PIN. Screen release does not send the report again. The delivery +coordinator owns the reporting backend session, so the new operation cannot +reuse or close it. + +A card with queued or in-flight reports immediately uses its protected snapshot +and locally reduced allowance, even when the general cache lacks its tanks. +Different cards authorize normally. Late delivery changes only its matching +report and attempt IDs. Network failures schedule retries; definitive rejections +move the unchanged payload to `dead_messages` and resolve that report. Once no +pending reports remain for the card, its next operation attempts online +authorization normally. + +Authorization has a bounded worker and queue. A separate single delivery worker +coordinates both new reports and retries, with one active sender and ordered +delivery within each card. At most one additional original session is retained; +other queued reports obtain an independent session when selected. Controller +state and display transitions remain on the controller loop. Shutdown cancels +active network work and joins the owned workers. At process exit the shared +token-cleanup worker is also cancelled and joined before DNS/logging teardown. +The same cleanup order applies when the application enters permanent failure. + +Controller startup and shutdown synchronize event-loop ownership, including a +thread that has been launched but has not entered `run()` yet. Exceptions release +that ownership too. `shutdown()` returns `false` if an active controller action +does not exit within `kShutdownDeadline`; it leaves live state and peripherals +intact so the caller can retry after the action exits. Destruction while the loop +is still live terminates the process. The application also exits with failure on +a shutdown timeout, allowing the service supervisor to restart it instead of +freeing state under a hung peripheral call. Independent/cancellable backend +sessions are required by `Controller` and checked at construction; synchronous +backend adapters may still be used directly by synchronization/delivery callers. + +## Persistence and compatibility + +The `backlog` schema now has stable increasing report IDs, delivery attempt IDs, +queued/in-flight status, retry eligibility, and a payload interpretation flag. +Startup migrates legacy implicit rowids in order without changing payloads. +Legacy payloads retain their existing tank-mapping interpretation. New payloads +capture the backend tank ID and transaction timestamp once; retries do not +remap tanks or replace timestamps. + +`card_report_state` stores the reduced allowance and saved user/tanks separately +from cache generations. Report acknowledgements and retries never deduct again. +Interrupted deliveries are recovered for retry at coordinator startup. Protected +card data and pending-card restrictions therefore survive restart. +If startup recovery cannot write to SQLite, the worker retries recovery before +sending any report. Recovery clears the in-flight marker; only the next delivery +claim increments the attempt ID. Old completions are rejected both before that +claim and after it. + +Canonical delivery checks the method's user role and payload shape. It does not +compare an already-recorded transaction with a later server allowance or remap +its tank: those values may have changed since dispensing, and doing so would +incorrectly suppress or alter a retry. + +Synchronization records resolved snapshot revisions before fetching server +data and may release protection only if those revisions remain unchanged and +the card still has no pending reports after the cache commit. A server fetch +started while reports were pending cannot restore an older allowance. A fresh +successful online authorization refreshes resolved protection and its revision, +so an older synchronization fetch cannot undo that authorization either. + +Backend deduplication remains unverified. Network-failure retries are preserved. +A timeout can occur after the server accounted for a report but before its reply +arrived; stable local IDs and exclusive sending cannot prevent duplicate server +accounting in that case. Exactly-once accounting requires backend support for +an idempotency key or equivalent reconciliation protocol. The local report ID +is not sent as an undocumented backend field. + +## Verification + +`foreground_backend_test.cpp` exercises configurable waits, early/late results, +cache fallback, missing-cache cancellation, same/different-card continuation, +refuel/intake delivery, rejection, storage failures, and shutdown. +`report_delivery_test.cpp` covers atomic deductions, snapshot protection, stable +payloads, attempt matching, per-card ordering, migration, and restart recovery. +Backend tests include cancellation of an already-sent HTTP authorization; +c-ares cancellation tests are enabled in `USE_CARES` builds. Display tests check +the Russian warning and both Cancel labels against the small display's width. + +Run `ctest --test-dir --output-on-failure`. The Windows console +suite and Linux container build with `TARGET_SIM800C=ON` have been exercised. +The Linux checks include cancellation while waiting for a deliberately withheld +DNS reply and while waiting for another resolver call. Physical displays and +the modem still require validation on the target hardware. diff --git a/docs/state_machine.md b/docs/state_machine.md index 2584ded..598cf7a 100644 --- a/docs/state_machine.md +++ b/docs/state_machine.md @@ -4,6 +4,12 @@ The FuelFlux controller uses a Mealy state machine to manage the fuel dispensing workflow. The state machine ensures proper sequencing of operations and handles various user interactions, timeouts, and error conditions. +Authorization and report delivery run on separate workers. The shared +`kForegroundBackendWaitTimeout` (default 5 seconds, rebuild to change) controls +cache fallback, the slow-connection warning, and release of transmission screens. +See [Foreground Backend Waits](foreground_backend_waits.md) for persistence, +pending-card continuation, late responses, and retry behavior. + ## States ### SystemState Enumeration @@ -42,12 +48,14 @@ The FuelFlux controller uses a Mealy state machine to manage the fuel dispensing | `PinEntered` | User completed PIN entry (pressed 'A') | | `AuthorizationSuccess` | Backend authorization succeeded | | `AuthorizationFailed` | Backend authorization failed | +| `AuthorizationCancelled` | User abandoned a pending attempt from the slow-connection screen | | `TankSelected` | User selected a valid tank | | `VolumeEntered` | User entered a valid volume | | `AmountEntered` | User entered a payment amount (future use) | | `RefuelingStarted` | Pump started dispensing fuel | | `RefuelingStopped` | Pump stopped (manually or target reached) | -| `DataTransmissionComplete` | Backend transaction transmission completed | +| `DataTransmissionComplete` | Report resolved or foreground wait expired after durable storage | +| `FlowDisplayRefresh` | Refresh the meter display without treating it as keyboard input | | `IntakeSelected` | Operator selected intake operation | | `IntakeDirectionSelected` | Operator selected intake direction (In/Out) | | `IntakeVolumeEntered` | Operator entered intake volume | @@ -128,9 +136,10 @@ reporting, offline backlog storage, and allowance deduction. - Both trigger: Event: `RefuelingStopped` → State: `RefuelDataTransmission` 6. **Refuel Data Transmission** - - Transaction is logged to backend asynchronously + - Report and reduced local allowance are committed together before another operation is enabled + - The delivery coordinator sends the report using its own session - Display shows: "Data transmission in progress" - - User cannot interact during transmission + - At delivery completion or the configured foreground deadline, the screen is released; delivery/retries can continue in the background - Event: `DataTransmissionComplete` → State: `RefuelingComplete` 7. **Refueling Complete** @@ -200,9 +209,9 @@ reporting, offline backlog storage, and allowance deduction. - Volume to transfer - Direction (In: 1, Out: 2) - Timestamp - - Transaction is logged to backend asynchronously + - Transaction and saved session data are persisted locally, then sent by the delivery coordinator - Display shows: "Data transmission in progress" - - Operator cannot interact during transmission (all keys disabled) + - At delivery completion or the configured foreground deadline, the screen is released; delivery/retries can continue in the background - Event: `DataTransmissionComplete` → State: `IntakeComplete` 7. **Intake Complete** @@ -266,7 +275,9 @@ When timeout occurs during intake entry: session ends, returns to `Waiting`, ope |----------|-------|--------|--------| | Invalid direction (e.g., '3') | `IntakeDirectionSelection` | Shows error, clears input | Stay in `IntakeDirectionSelection` | | Zero or negative volume | `IntakeVolumeEntry` | Shows error, clears input | Stay in `IntakeVolumeEntry` | -| Backend transmission fails | `IntakeDataTransmission` | Transaction not logged | Stay in `IntakeDataTransmission`, can retry with cancel+restart | +| Network transmission fails | `IntakeDataTransmission` | Durable report queued for retry | Completion screen after foreground deadline; same card uses saved data | +| Backend definitively rejects report | `IntakeDataTransmission` | Move report to failed reports | Report resolved; next online authorization allowed when card has no pending reports | +| Local report storage fails | `IntakeDataTransmission` | Show `Ошибка записи` | Enter `Error`; do not release an unrecorded transaction | | Invalid tank number | `TankSelection` | Shows error, clears input | Stay in `TankSelection` | | Timeout during intake | Any intake state | Session cleared | Return to `Waiting` | @@ -317,7 +328,7 @@ The 'B' key (Stop/Cancel) is active in most states and handles different operati |------------|--------| | `Waiting` | No effect | | `PinEntry` | Clear PIN, return to `Waiting` | -| `Authorization` | No effect (cannot cancel during auth) | +| `Authorization` | No effect before foreground deadline; on the slow-connection screen, abandon attempt and return to `Waiting` | | `NotAuthorized` | Return to `Waiting` | | `TankSelection` | Cancel selection, return to `Waiting` | | `VolumeEntry` | Cancel entry, return to `Waiting` | diff --git a/include/authorization_snapshot.h b/include/authorization_snapshot.h new file mode 100644 index 0000000..d463a5b --- /dev/null +++ b/include/authorization_snapshot.h @@ -0,0 +1,25 @@ +#pragma once + +#include "backend.h" +#include + +namespace fuelflux { + +// A value copy: it must not change when synchronization flips cache tables. +struct AuthorizationSnapshot { + UserInfo user; + std::vector tanks; +}; + +struct ProtectedCardSnapshot { + AuthorizationSnapshot authorization; + bool pending = false; +}; + +struct SavedAuthorizationState { + std::optional saved; + bool reportStorageAvailable = false; + bool pendingReports = true; // Fail closed if storage cannot be read. +}; + +} // namespace fuelflux diff --git a/include/backend.h b/include/backend.h index 88f97d3..10dda86 100644 --- a/include/backend.h +++ b/include/backend.h @@ -49,6 +49,12 @@ struct FuelTank { class IBackend { public: virtual ~IBackend() = default; + // Return an independent cancellable session for each foreground operation. + // Empty is allowed for synchronous-only adapters. Controller requires this + // capability and rejects unsupported adapters at construction, before use. + virtual std::shared_ptr CreateIndependentSession() const { return {}; } + virtual void CancelPendingRequests() = 0; + virtual bool SendReportPayload(const std::string& payload, bool intake, bool canonicalTankId) = 0; virtual bool Authorize(const std::string& uid) = 0; virtual bool Deauthorize() = 0; virtual bool Refuel(TankNumber tankNumber, Volume volume) = 0; @@ -86,6 +92,9 @@ class IBackend { class BackendBase : public IBackend, public std::enable_shared_from_this { public: ~BackendBase() override = default; + // Process shutdown only: cancel and join the shared token-cleanup worker + // before tearing down DNS and logging. Independent controller workers stop first. + static void ShutdownAsyncRequests(); bool Authorize(const std::string& uid) override; bool Deauthorize() override; @@ -93,6 +102,8 @@ class BackendBase : public IBackend, public std::enable_shared_from_this& DeauthorizeCancellation(); // Map TankNumber in a payload from visualNumberTank to idTank. // Returns true if a match was found and the mapping was applied. @@ -140,6 +152,7 @@ class BackendBase : public IBackend, public std::enable_shared_from_this fuelTanks_; std::string lastError_; std::atomic networkError_{false}; + std::atomic cancelled_{false}; std::shared_ptr storage_; }; @@ -153,6 +166,7 @@ class Backend : public BackendBase { Backend(const std::string& baseAPI, const std::string& controllerUid, std::shared_ptr storage = nullptr); ~Backend() override; + std::shared_ptr CreateIndependentSession() const override; private: // Private method for common parsing of responses from the backend diff --git a/include/backlog_worker.h b/include/backlog_worker.h index fbaf8d5..5b7756e 100644 --- a/include/backlog_worker.h +++ b/include/backlog_worker.h @@ -10,6 +10,8 @@ #include #include #include +#include +#include #include "backend.h" #include "message_storage.h" @@ -33,11 +35,16 @@ class BacklogWorker { void SetInterval(std::chrono::milliseconds interval); bool ProcessOnce(); + void Wake(); + // Atomically submit a report and reserve its original session for delivery. + std::optional Submit(MessageMethod method, const std::string& payload, + AuthorizationSnapshot snapshot, double deduction, std::shared_ptr session = nullptr); private: void RunLoop(); bool ProcessMessage(const StoredMessage& message); bool HandleFailure(const StoredMessage& message); + bool FinishDelivery(const StoredMessage& message, MessageStorage::DeliveryResult result); std::shared_ptr storage_; std::shared_ptr backend_; @@ -46,6 +53,11 @@ class BacklogWorker { std::thread workerThread_; mutable std::mutex mutex_; std::condition_variable cv_; + bool wake_ = false; + std::mutex deliveryMutex_; + std::unordered_map> sessions_; + std::shared_ptr activeBackend_; + bool activeSessionSupplied_ = false; }; } // namespace fuelflux diff --git a/include/cache_manager.h b/include/cache_manager.h index fb66c03..de4d6f6 100644 --- a/include/cache_manager.h +++ b/include/cache_manager.h @@ -7,6 +7,7 @@ #include "user_cache.h" #include "backend.h" #include "timing_config.h" +#include "message_storage.h" #include #include #include @@ -52,6 +53,9 @@ class CacheManager { // Deduct allowance from cache (called after refuel for RoleId==1) bool DeductAllowance(const std::string& uid, double amount); + // Configure before Start(). Only a fresh, complete population may release + // resolved local snapshots; requests begun before delivery must not do so. + void SetReportStorage(std::shared_ptr storage) { reportStorage_ = std::move(storage); } private: void WorkerThread(); @@ -60,6 +64,7 @@ class CacheManager { std::shared_ptr cache_; std::shared_ptr backend_; + std::shared_ptr reportStorage_; std::thread workerThread_; std::atomic running_; diff --git a/include/cares_resolver.h b/include/cares_resolver.h index 9d665c5..ad6073c 100644 --- a/include/cares_resolver.h +++ b/include/cares_resolver.h @@ -11,6 +11,7 @@ #include #include #include +#include #include "timing_config.h" namespace fuelflux { @@ -40,7 +41,8 @@ class CaresResolver { using TimeProvider = std::function; CaresResolver(); - explicit CaresResolver(const std::string& cachedHostname, TimeProvider timeProvider = Clock::now); + explicit CaresResolver(const std::string& cachedHostname, TimeProvider timeProvider = Clock::now, + const std::string& dnsServers = ""); ~CaresResolver(); // Resolve hostname to IP address using Yandex DNS @@ -50,7 +52,8 @@ class CaresResolver { // interface - network interface to bind for DNS queries (e.g., "ppp0") // empty string means use system default // Thread-safe: multiple threads can call this method concurrently - std::string Resolve(const std::string& hostname, const std::string& interface = ""); + std::string Resolve(const std::string& hostname, const std::string& interface = "", + const std::atomic* cancelled = nullptr); // Test helpers for validating targeted cache behavior bool HasValidTargetedCacheForTesting() const; @@ -69,11 +72,12 @@ class CaresResolver { std::string cached_hostname_; TimeProvider time_provider_; + std::string dns_servers_; std::optional backend_api_cache_entry_; // Mutex for thread-safe DNS resolution // Protects concurrent channel operations - mutable std::mutex resolve_mutex_; + mutable std::timed_mutex resolve_mutex_; }; } // namespace fuelflux diff --git a/include/controller.h b/include/controller.h index 3cbb54c..d5a0555 100644 --- a/include/controller.h +++ b/include/controller.h @@ -17,6 +17,7 @@ #include "backend.h" #include "message_storage.h" +#include "backlog_worker.h" #include "state_machine.h" #include "timing_config.h" #include "types.h" @@ -42,12 +43,15 @@ class Controller { Controller(ControllerId controllerId, std::shared_ptr backend, std::chrono::seconds noFlowCancelTimeout, - ControllerPersistencePaths persistencePaths); + ControllerPersistencePaths persistencePaths, + std::chrono::milliseconds foregroundWait = timing::kForegroundBackendWaitTimeout); ~Controller(); // System lifecycle bool initialize(); - void shutdown(); + // False means the event loop missed its deadline. Keep this object and its + // peripherals alive, then retry/join after it exits or terminate the process. + bool shutdown(); void run(); /** * Reinitialize the device and all connected peripherals. @@ -132,7 +136,11 @@ class Controller { void setMaxValue(); // Authorization + // Synchronous compatibility helper; foreground state transitions use beginAuthorization. void requestAuthorization(const UserId& userId); + void beginAuthorization(const UserId& userId); + bool isAuthorizationSlow() const { return authorizationSlow_.load(); } + void cancelSlowAuthorization(); // Tank operations void selectTank(TankNumber tankNumber); @@ -191,7 +199,30 @@ class Controller { std::unique_ptr temperatureSensor_; std::unique_ptr gpsReceiver_; std::shared_ptr backend_; + std::shared_ptr backendPrototype_; std::shared_ptr messageStorage_; + std::unique_ptr reportWorker_; + std::chrono::milliseconds foregroundWait_; + BoundedExecutor authorizationExecutor_{1, 1}; + struct AuthorizationAttempt { + unsigned long long id = 0; + std::string uid; + std::shared_ptr backend; + std::optional saved; + AuthorizationSnapshot online; + std::chrono::steady_clock::time_point started; + std::atomic done{false}; + bool success = false; + bool networkError = false; + bool adopted = false; + ~AuthorizationAttempt(); + }; + std::shared_ptr authorizationAttempt_; + unsigned long long nextAuthorizationId_ = 0; + std::atomic authorizationSlow_{false}; + std::atomic activeAuthorizationId_{0}; + std::optional foregroundReport_; + std::chrono::steady_clock::time_point reportStarted_; // Cache components std::shared_ptr userCache_; @@ -225,13 +256,24 @@ class Controller { bool stopPressBeganInWaiting_ = false; // System state - bool isRunning_; - std::atomic threadExited_{false}; + std::atomic isRunning_; + std::mutex shutdownMutex_; + std::mutex lifecycleMutex_; + std::condition_variable lifecycleCv_; + bool loopActive_ = false; + bool shutdownRequested_ = false; + bool peripheralsNeedShutdown_ = false; std::string lastErrorMessage_; bool sessionAuthorizedFromCache_ = false; // Event queue for cross-thread event posting - std::queue eventQueue_; + struct QueuedEvent { + Event event; + unsigned long long authorizationId = 0; + bool authorizationCancel = false; + bool cancelEnabled = false; + }; + std::queue eventQueue_; std::mutex eventQueueMutex_; std::condition_variable eventCv_; @@ -262,6 +304,14 @@ class Controller { Volume parseVolumeFromInput() const; TankNumber parseTankFromInput() const; void resetSessionData(); + SavedAuthorizationState savedAuthorization(const std::string& uid) const; + void applyAuthorization(const AuthorizationSnapshot& snapshot, bool fromCache); + void pollBackendOperations(); + void finishRun(); + void abandonAuthorization(); + bool submitReport(MessageMethod method, const std::string& uid, TankNumber tank, + Volume volume, std::chrono::system_clock::time_point timestamp, + IntakeDirection direction = IntakeDirection::In); void selectIntakeDirection(IntakeDirection direction); /** * Initializes all configured peripherals (display, keyboard, card reader, pump, flow meter, backend). diff --git a/include/message_storage.h b/include/message_storage.h index 39c0fa9..60c0102 100644 --- a/include/message_storage.h +++ b/include/message_storage.h @@ -8,6 +8,9 @@ #include #include #include +#include +#include +#include "authorization_snapshot.h" struct sqlite3; @@ -23,6 +26,8 @@ struct StoredMessage { std::string uid; MessageMethod method = MessageMethod::Refuel; std::string data; + long long attempt = 0; + bool canonicalTankId = false; }; class MessageStorage { @@ -34,6 +39,10 @@ class MessageStorage { MessageStorage& operator=(const MessageStorage&) = delete; bool IsOpen() const; + // Capture pending-card protection and storage availability under one SQLite + // transaction. The callback copies a single general-cache generation. + SavedAuthorizationState CaptureAuthorizationState(const std::string& uid, + const std::function()>& generalCache) const; bool AddBacklog(const std::string& uid, MessageMethod method, const std::string& data); bool AddDeadMessage(const std::string& uid, MessageMethod method, const std::string& data); @@ -41,6 +50,21 @@ class MessageStorage { std::optional GetNextBacklog(); bool RemoveBacklog(long long id); + // Reports and their local allowance are committed together before UI release. + std::optional EnqueueReport(MessageMethod method, const std::string& data, + AuthorizationSnapshot snapshot, double deduction); + std::optional GetProtectedSnapshot(const std::string& uid) const; + bool ClearProtectedSnapshot(const std::string& uid); + bool RefreshResolvedSnapshot(const AuthorizationSnapshot& snapshot); + std::vector> ResolvedSnapshotVersions() const; + void ReleaseResolvedSnapshots(const std::vector>& versions); + bool HasPendingReports(const std::string& uid) const; + bool HasReport(long long id) const; + std::optional ClaimNextBacklog(); + enum class DeliveryResult { Accepted, Retry, Rejected }; + bool CompleteDelivery(const StoredMessage& message, DeliveryResult result, int retrySeconds = 30); + bool RecoverInFlight(); + int BacklogCount() const; int DeadMessageCount() const; diff --git a/include/timing_config.h b/include/timing_config.h index c1b704e..e7a18f4 100644 --- a/include/timing_config.h +++ b/include/timing_config.h @@ -30,6 +30,12 @@ constexpr std::chrono::seconds kCalibrationSavedDisplayDuration{2}; // ─── Controller ─────────────────────────────────────────────────────────────── +// Maximum foreground wait before cache fallback or background report delivery. +// Transport deadlines remain independent of this user-interface threshold. +constexpr std::chrono::seconds kForegroundBackendWaitTimeout{5}; +static_assert(kForegroundBackendWaitTimeout.count() > 0, + "Foreground backend wait must be positive"); + // Default no-flow cancel timeout: pump stops if no flow pulses are received // for this long during refuelling (seconds). constexpr std::chrono::seconds kNoFlowCancelTimeout{30}; @@ -42,7 +48,7 @@ constexpr std::chrono::milliseconds kEventLoopWaitInterval{100}; // without an event (avoids busy-spinning). constexpr std::chrono::milliseconds kEventLoopIdleSleep{10}; -// Shutdown: maximum time to wait for the event-loop thread to exit. +// Shutdown: maximum wait for the event-loop thread before reporting failure. constexpr std::chrono::milliseconds kShutdownDeadline{2000}; // No-flow monitor thread: polling interval. diff --git a/include/types.h b/include/types.h index 46b9d6c..e4f0c63 100644 --- a/include/types.h +++ b/include/types.h @@ -67,6 +67,7 @@ enum class Event { AuthorizationSuccess, AuthorizationDenied, AuthorizationFailed, + AuthorizationCancelled, TankSelected, VolumeEntered, AmountEntered, @@ -82,7 +83,8 @@ enum class Event { Timeout, Error, ErrorRecovery, - DisplayReset + DisplayReset, + FlowDisplayRefresh }; // Key codes for keyboard input diff --git a/include/user_cache.h b/include/user_cache.h index 5b8209f..3e68a6e 100644 --- a/include/user_cache.h +++ b/include/user_cache.h @@ -8,6 +8,7 @@ #include #include #include +#include "authorization_snapshot.h" struct sqlite3; @@ -40,6 +41,7 @@ class UserCache { // Cache operations std::optional GetEntry(const std::string& uid) const; + std::optional GetAuthorizationSnapshot(const std::string& uid) const; bool UpdateEntry(const std::string& uid, double allowance, int roleId); bool DeductAllowance(const std::string& uid, double amount); int GetCount() const; @@ -60,7 +62,7 @@ class UserCache { sqlite3* db_; std::string dbPath_; - mutable std::mutex dbMutex_; + mutable std::recursive_mutex dbMutex_; bool activeTableIsA_; // true = table A is active, false = table B is active bool populationInProgress_; }; diff --git a/src/backend.cpp b/src/backend.cpp index 880b754..ac7b17b 100644 --- a/src/backend.cpp +++ b/src/backend.cpp @@ -286,8 +286,10 @@ CaresResolver& GetCaresResolver() { // logPrefix - prefix for log messages (e.g., "" or "Async deauthorize: ") void SetupDnsResolution(CURL* curl, CurlSlist& resolveList, const std::string& host, const std::string& url, - const std::string& logPrefix = "") { - std::string resolvedIp = GetCaresResolver().Resolve(host, kPppInterface); + const std::string& logPrefix = "", + const std::atomic* cancelled = nullptr) { + std::string resolvedIp = GetCaresResolver().Resolve(host, kPppInterface, cancelled); + if (cancelled && cancelled->load()) return; if (resolvedIp.empty()) { // c-ares failed; fall back to letting libcurl do DNS, but force it to use // the PPP interface so that queries do not go via the default route. @@ -325,6 +327,11 @@ Backend::~Backend() { // Cleanup should happen at process shutdown, not per Backend instance } +std::shared_ptr Backend::CreateIndependentSession() const { + // Delivery persistence belongs to the coordinator, not the transport. + return std::make_shared(baseAPI_, controllerUid_, nullptr); +} + nlohmann::json Backend::HttpRequestWrapper(const std::string& endpoint, const std::string& method, const nlohmann::json& requestBody, @@ -333,6 +340,11 @@ nlohmann::json Backend::HttpRequestWrapper(const std::string& endpoint, networkError_ = false; + if (cancelled_.load()) { + networkError_ = true; + return BuildWrapperErrorResponse(); + } + // Use RAII wrapper for CURL handle CurlHandle curl; if (!curl) { @@ -358,6 +370,12 @@ nlohmann::json Backend::HttpRequestWrapper(const std::string& endpoint, // Set callback for response curl_easy_setopt(curl.get(), CURLOPT_WRITEFUNCTION, WriteCallback); curl_easy_setopt(curl.get(), CURLOPT_WRITEDATA, &responseBody); + curl_easy_setopt(curl.get(), CURLOPT_NOPROGRESS, 0L); + curl_easy_setopt(curl.get(), CURLOPT_XFERINFODATA, &cancelled_); + curl_easy_setopt(curl.get(), CURLOPT_XFERINFOFUNCTION, + +[](void* context, curl_off_t, curl_off_t, curl_off_t, curl_off_t) -> int { + return static_cast*>(context)->load() ? 1 : 0; + }); // Set timeouts #ifdef TARGET_SIM800C @@ -389,7 +407,7 @@ nlohmann::json Backend::HttpRequestWrapper(const std::string& endpoint, // Bind DNS queries to ppp0 interface via ares_set_local_dev() // Then use CURLOPT_RESOLVE to provide the resolved IP to curl // This preserves the hostname in the URL for Host header and SNI - SetupDnsResolution(curl.get(), resolveList, host, url); + SetupDnsResolution(curl.get(), resolveList, host, url, "", &cancelled_); #endif } else { if (IsLocalhost(host)) { @@ -429,6 +447,10 @@ nlohmann::json Backend::HttpRequestWrapper(const std::string& endpoint, } // Perform request + if (cancelled_.load()) { + networkError_ = true; + return BuildWrapperErrorResponse(); + } CURLcode res = curl_easy_perform(curl.get()); if (res != CURLE_OK) { @@ -641,6 +663,8 @@ nlohmann::json Backend::HttpRequestWrapper(const std::string& endpoint, // Static helper for async deauthorize - doesn't use mutex or modify state void Backend::SendAsyncDeauthorize(const std::string& baseAPI, const std::string& token) { + auto& cancelled = DeauthorizeCancellation(); + if (cancelled.load()) return; // Use RAII wrapper for CURL handle CurlHandle curl; if (!curl) { @@ -655,6 +679,13 @@ void Backend::SendAsyncDeauthorize(const std::string& baseAPI, const std::string LOG_BCK_DEBUG("Async deauthorize: POST /api/pump/deauthorize"); + curl_easy_setopt(curl.get(), CURLOPT_NOPROGRESS, 0L); + curl_easy_setopt(curl.get(), CURLOPT_XFERINFODATA, &cancelled); + curl_easy_setopt(curl.get(), CURLOPT_XFERINFOFUNCTION, + +[](void* context, curl_off_t, curl_off_t, curl_off_t, curl_off_t) -> int { + return static_cast*>(context)->load() ? 1 : 0; + }); + // Set URL curl_easy_setopt(curl.get(), CURLOPT_URL, url.c_str()); @@ -690,7 +721,7 @@ void Backend::SendAsyncDeauthorize(const std::string& baseAPI, const std::string #ifdef USE_CARES // Use c-ares with Yandex DNS for hostname resolution via ppp0 - SetupDnsResolution(curl.get(), resolveList, host, url, "Async deauthorize: "); + SetupDnsResolution(curl.get(), resolveList, host, url, "Async deauthorize: ", &cancelled); #endif } #endif @@ -713,6 +744,7 @@ void Backend::SendAsyncDeauthorize(const std::string& baseAPI, const std::string curl_easy_setopt(curl.get(), CURLOPT_POSTFIELDSIZE, static_cast(bodyStr.size())); // Perform request + if (cancelled.load()) return; CURLcode res = curl_easy_perform(curl.get()); if (res != CURLE_OK) { diff --git a/src/backend_base.cpp b/src/backend_base.cpp index 5300e89..e683457 100644 --- a/src/backend_base.cpp +++ b/src/backend_base.cpp @@ -6,6 +6,7 @@ #include #include +#include #include #include @@ -98,14 +99,27 @@ bool ParseFuelTank(const nlohmann::json& tank, } // namespace -// Meyer's singleton for bounded executor - thread-safe lazy initialization -// Initialized on first use, avoiding static initialization order issues -// and allowing exception handling at runtime instead of during startup +namespace { +struct DeauthorizeWorker { + std::atomic cancelled{false}; + BoundedExecutor executor{1, 100}; + void Stop() { cancelled.store(true); executor.Shutdown(); } + ~DeauthorizeWorker() { Stop(); } +}; + +DeauthorizeWorker& deauthorizeWorker() { + static DeauthorizeWorker worker; + return worker; +} +} // namespace + BoundedExecutor& BackendBase::GetDeauthorizeExecutor() { - static BoundedExecutor executor(1, 100); - return executor; + return deauthorizeWorker().executor; } +std::atomic& BackendBase::DeauthorizeCancellation() { return deauthorizeWorker().cancelled; } +void BackendBase::ShutdownAsyncRequests() { deauthorizeWorker().Stop(); } + BackendBase::BackendBase(std::string controllerUid, std::shared_ptr storage) : controllerUid_(std::move(controllerUid)) , storage_(std::move(storage)) @@ -236,16 +250,16 @@ bool BackendBase::Deauthorize() { // Try to submit async HTTP request to bounded executor if backend is managed by shared_ptr // Uses a dedicated async method that doesn't hold requestMutex_ or modify networkError_ try { - std::weak_ptr weakSelf = shared_from_this(); - bool submitted = GetDeauthorizeExecutor().Submit([weakSelf, token]() { - // Check if backend still exists - if (auto self = weakSelf.lock()) { - try { - // Call virtual method that sends request without mutex - self->SendAsyncDeauthorizeRequest(token); - } catch (const std::exception& e) { - LOG_BCK_WARN("Async deauthorization failed (ignored): {}", e.what()); - } + auto self = shared_from_this(); + // Keep the independent session alive until token cleanup completes. + // The executor's fixed queue bounds this lifetime extension. + bool submitted = GetDeauthorizeExecutor().Submit([self, token]() { + if (DeauthorizeCancellation().load()) return; + try { + // Call virtual method that sends request without mutex + self->SendAsyncDeauthorizeRequest(token); + } catch (const std::exception& e) { + LOG_BCK_WARN("Async deauthorization failed (ignored): {}", e.what()); } }); @@ -494,6 +508,34 @@ bool BackendBase::RefuelPayload(const std::string& payload) { } } +bool BackendBase::SendReportPayload(const std::string& payload, bool intake, bool canonicalTankId) { + if (!canonicalTankId) return intake ? IntakePayload(payload) : RefuelPayload(payload); + networkError_ = false; + const auto requiredRole = intake ? UserRole::Operator : UserRole::Customer; + if (!session_.IsAuthorized() || roleId_ != static_cast(requiredRole)) { + lastError_ = StdControllerError; + return false; + } + const auto body = nlohmann::json::parse(payload, nullptr, false); + const char* volumeKey = intake ? "IntakeVolume" : "FuelVolume"; + if (!body.is_object() || !body.contains(volumeKey) || !body[volumeKey].is_number() || + !std::isfinite(body[volumeKey].get()) || body[volumeKey].get() < 0 || + !body.contains("TankNumber") || !body["TankNumber"].is_number_integer() || + !body.contains("TimeAt") || !body["TimeAt"].is_number_integer() || + (intake && (!body.contains("Direction") || !body["Direction"].is_number_integer() || + (body["Direction"] != static_cast(IntakeDirection::In) && + body["Direction"] != static_cast(IntakeDirection::Out))))) { + lastError_ = StdControllerError; + return false; + } + // This is delivery of an already-recorded transaction. A later server + // allowance or tank mapping must not suppress/remap its retry. + const auto response = HttpRequestWrapper(intake ? "/api/pump/fuel-intake" : "/api/pump/refuel", "POST", body, true); + if (IsErrorResponse(response, &lastError_)) return false; + lastError_.clear(); + return true; +} + bool BackendBase::IntakePayload(const std::string& payload) { try { if (!session_.IsAuthorized()) { diff --git a/src/backlog_worker.cpp b/src/backlog_worker.cpp index 618e63b..e11e8ea 100644 --- a/src/backlog_worker.cpp +++ b/src/backlog_worker.cpp @@ -32,9 +32,16 @@ void BacklogWorker::Stop() { return; } cv_.notify_all(); + { + std::lock_guard lock(mutex_); + if (activeBackend_) activeBackend_->CancelPendingRequests(); + } if (workerThread_.joinable()) { workerThread_.join(); } + std::lock_guard lock(mutex_); + for (auto& entry : sessions_) entry.second->Deauthorize(); + sessions_.clear(); } bool BacklogWorker::IsRunning() const { @@ -50,65 +57,131 @@ void BacklogWorker::SetInterval(std::chrono::milliseconds interval) { } void BacklogWorker::RunLoop() { + bool recovered = false; while (running_.load()) { - const bool processed = ProcessOnce(); + // No normal delivery until interrupted claims have been recovered. + // A transient SQLite error must not strand them until the next restart. + if (!recovered) recovered = storage_ && storage_->RecoverInFlight(); + const bool processed = recovered && running_.load() && ProcessOnce(); std::unique_lock lock(mutex_); if (!running_.load()) { break; } if (!processed) { - cv_.wait_for(lock, interval_, [this] { return !running_.load(); }); + cv_.wait_for(lock, interval_, [this] { return !running_.load() || wake_; }); + wake_ = false; } } } bool BacklogWorker::ProcessOnce() { + std::lock_guard deliveryLock(deliveryMutex_); if (!storage_ || !backend_) { return false; } - const auto message = storage_->GetNextBacklog(); + std::optional message; + { + std::lock_guard lock(mutex_); + if (workerThread_.joinable() && !running_.load()) { + return false; + } + message = storage_->ClaimNextBacklog(); + if (message) { + const auto it = sessions_.find(message->id); + activeSessionSupplied_ = it != sessions_.end(); + if (it != sessions_.end()) { + activeBackend_ = std::move(it->second); + sessions_.erase(it); + } else { + activeBackend_ = backend_->CreateIndependentSession(); + if (!activeBackend_) activeBackend_ = backend_; + } + } + } if (!message) { return false; } - return ProcessMessage(*message); + bool result = false; + try { result = ProcessMessage(*message); } + catch (const std::exception& e) { + LOG_BCK_ERROR("Report {} delivery exception: {}", message->id, e.what()); + FinishDelivery(*message, MessageStorage::DeliveryResult::Retry); + if (activeBackend_->IsAuthorized()) activeBackend_->Deauthorize(); + } + { + std::lock_guard lock(mutex_); + activeBackend_.reset(); + } + return result; +} + +void BacklogWorker::Wake() { + { std::lock_guard lock(mutex_); wake_ = true; } + cv_.notify_one(); +} + +std::optional BacklogWorker::Submit(MessageMethod method, const std::string& payload, + AuthorizationSnapshot snapshot, double deduction, std::shared_ptr session) { + std::optional id; + { + std::lock_guard lock(mutex_); + id = storage_->EnqueueReport(method, payload, std::move(snapshot), deduction); + // Retain at most one waiting session. Other durable reports obtain a + // fresh session when selected, rather than building an unbounded queue. + if (id && session && sessions_.empty()) sessions_.emplace(*id, std::move(session)); + if (id) wake_ = true; + } + if (id && session) session->Deauthorize(); + cv_.notify_one(); + return id; } bool BacklogWorker::HandleFailure(const StoredMessage& message) { - if (backend_->IsNetworkError()) { + if (activeBackend_->IsNetworkError()) { LOG_BCK_WARN("Backlog processing paused due to network error"); + FinishDelivery(message, MessageStorage::DeliveryResult::Retry); return false; } LOG_BCK_WARN("Moving backlog message {} to dead messages", message.id); - storage_->AddDeadMessage(message.uid, message.method, message.data); - storage_->RemoveBacklog(message.id); + return FinishDelivery(message, MessageStorage::DeliveryResult::Rejected); +} + +bool BacklogWorker::FinishDelivery(const StoredMessage& message, MessageStorage::DeliveryResult result) { + int retrySeconds; + { + std::lock_guard lock(mutex_); + retrySeconds = static_cast(std::chrono::duration_cast(interval_).count()); + } + // A temporary SQLite failure must not strand a live claim or cause another + // network send. Keep the completed outcome until it can be committed. + while (!storage_->CompleteDelivery(message, result, retrySeconds)) { + if (!running_.load()) return false; + std::unique_lock lock(mutex_); + cv_.wait_for(lock, std::chrono::milliseconds(100), [this] { return !running_.load(); }); + } return true; } bool BacklogWorker::ProcessMessage(const StoredMessage& message) { - if (!backend_->Authorize(message.uid)) { + if (!activeSessionSupplied_ && !activeBackend_->Authorize(message.uid)) { return HandleFailure(message); } - bool sendOk = false; - if (message.method == MessageMethod::Refuel) { - sendOk = backend_->RefuelPayload(message.data); - } else { - sendOk = backend_->IntakePayload(message.data); - } + const bool sendOk = activeBackend_->SendReportPayload(message.data, + message.method == MessageMethod::Intake, message.canonicalTankId); // Deauthorize is treated as fire-and-forget at this call site. // Return value and potential errors are intentionally ignored. - backend_->Deauthorize(); + activeBackend_->Deauthorize(); if (!sendOk) { return HandleFailure(message); } - storage_->RemoveBacklog(message.id); - return true; + return FinishDelivery(message, MessageStorage::DeliveryResult::Accepted); } } // namespace fuelflux diff --git a/src/cache_manager.cpp b/src/cache_manager.cpp index 17cc7d5..8f8e15b 100644 --- a/src/cache_manager.cpp +++ b/src/cache_manager.cpp @@ -27,6 +27,9 @@ bool CacheManager::Start() { return false; } + if (backend_) { + if (auto session = backend_->CreateIndependentSession()) backend_ = std::move(session); + } running_ = true; workerThread_ = std::thread(&CacheManager::WorkerThread, this); @@ -38,6 +41,7 @@ bool CacheManager::Start() { } void CacheManager::Stop() { + if (backend_) backend_->CancelPendingRequests(); if (!running_) { return; } @@ -174,6 +178,8 @@ void CacheManager::WorkerThread() { } bool CacheManager::PopulateCache() { + const auto resolvedVersions = reportStorage_ ? reportStorage_->ResolvedSnapshotVersions() + : std::vector>{}; if (!cache_ || !backend_) { LOG_ERROR("Cache or backend not available"); return false; @@ -302,6 +308,7 @@ bool CacheManager::PopulateCache() { // Close synchronization session - this is critical for cleanup // If deauthorization fails, we still return true because the data was successfully loaded + if (reportStorage_) reportStorage_->ReleaseResolvedSnapshots(resolvedVersions); // The backend will clean up the session automatically after timeout if (!backend_->Deauthorize()) { LOG_WARN("Failed to deauthorize synchronization session (data was still loaded successfully, backend will clean up on timeout)"); diff --git a/src/cares_resolver.cpp b/src/cares_resolver.cpp index 3150db7..eea0edc 100644 --- a/src/cares_resolver.cpp +++ b/src/cares_resolver.cpp @@ -39,7 +39,7 @@ std::condition_variable g_cares_init_cv; // RAII wrapper for ares_channel class AresChannel { public: - AresChannel(const std::string& interface) : channel_(nullptr), initialized_(false) { + AresChannel(const std::string& interface, const std::string& dnsServers) : channel_(nullptr), initialized_(false) { // Check if library is initialized if (g_cares_init_state.load(std::memory_order_acquire) != InitState::Initialized) { LOG_BCK_ERROR("c-ares library not initialized. Call InitializeCaresLibrary() first."); @@ -57,8 +57,9 @@ class AresChannel { return; } - // Set Yandex DNS servers - std::string dnsServers = std::string(kYandexDns1) + "," + kYandexDns2; + // Use the legacy c-ares API on purpose for broad platform compatibility. + // Defaults to Yandex DNS. An explicit server list also permits local + // resolver tests without relying on public DNS timing. status = ares_set_servers_csv(channel_, dnsServers.c_str()); if (status != ARES_SUCCESS) { LOG_BCK_ERROR("Failed to set DNS servers '{}': {} (error code: {})", @@ -210,20 +211,22 @@ void CleanupCaresLibrary() { CaresResolver::CaresResolver() : CaresResolver(ExtractHostFromUrl(BACKEND_API_URL), Clock::now) {} -CaresResolver::CaresResolver(const std::string& cachedHostname, TimeProvider timeProvider) +CaresResolver::CaresResolver(const std::string& cachedHostname, TimeProvider timeProvider, + const std::string& dnsServers) : cached_hostname_(cachedHostname), - time_provider_(std::move(timeProvider)) {} + time_provider_(std::move(timeProvider)), + dns_servers_(dnsServers.empty() ? std::string(kYandexDns1) + "," + kYandexDns2 : dnsServers) {} CaresResolver::~CaresResolver() { } bool CaresResolver::HasValidTargetedCacheForTesting() const { - std::lock_guard lock(resolve_mutex_); + std::lock_guard lock(resolve_mutex_); return HasValidBackendCacheEntry(); } std::string CaresResolver::GetTargetedCachedIpForTesting() const { - std::lock_guard lock(resolve_mutex_); + std::lock_guard lock(resolve_mutex_); if (!backend_api_cache_entry_.has_value()) { return ""; } @@ -238,9 +241,14 @@ bool CaresResolver::HasValidBackendCacheEntry() const { return backend_api_cache_entry_.has_value() && time_provider_() < backend_api_cache_entry_->expiresAt; } -std::string CaresResolver::Resolve(const std::string& hostname, const std::string& interface) { +std::string CaresResolver::Resolve(const std::string& hostname, const std::string& interface, + const std::atomic* cancelled) { // Serialize concurrent calls to prevent issues with channel operations - std::lock_guard lock(resolve_mutex_); + std::unique_lock lock(resolve_mutex_, std::defer_lock); + while (!lock.try_lock_for(std::chrono::milliseconds(100))) { + if (cancelled && cancelled->load()) return ""; + } + if (cancelled && cancelled->load()) return ""; // Check if library is initialized // NOTE: This check is for detecting programming errors (using resolver before init). @@ -270,14 +278,16 @@ std::string CaresResolver::Resolve(const std::string& hostname, const std::strin return backend_api_cache_entry_->ip; } - AresChannel channel(interface); + // ares_destroy invokes pending callbacks, so their context must outlive + // the channel even when cancellation exits this function early. + ResolveContext ctx; + AresChannel channel(interface, dns_servers_); if (!channel.isInitialized()) { LOG_BCK_ERROR("Failed to initialize c-ares channel"); return ""; } // Perform DNS resolution - ResolveContext ctx; ares_gethostbyname(channel.get(), hostname.c_str(), AF_INET, HostCallback, &ctx); // Wait for resolution to complete with overall timeout protection @@ -292,6 +302,7 @@ std::string CaresResolver::Resolve(const std::string& hostname, const std::strin int loopCount = 0; while (!ctx.done) { + if (cancelled && cancelled->load()) return ""; // Check overall timeout to prevent infinite hangs auto elapsed = std::chrono::duration_cast( std::chrono::steady_clock::now() - startTime); @@ -313,7 +324,8 @@ std::string CaresResolver::Resolve(const std::string& hostname, const std::strin // Set select timeout (max kDnsSelectTimeoutSec seconds per iteration) max_tv.tv_sec = timing::kDnsSelectTimeoutSec; - max_tv.tv_usec = 0; + if (cancelled) max_tv.tv_sec = 0; + max_tv.tv_usec = cancelled ? 100000 : 0; struct timeval* tvp = ares_timeout(channel.get(), &max_tv, &tv); int result = select(nfds, &read_fds, &write_fds, nullptr, tvp); diff --git a/src/controller.cpp b/src/controller.cpp index f88a083..8512007 100644 --- a/src/controller.cpp +++ b/src/controller.cpp @@ -1,4 +1,4 @@ -// Copyright (C) 2025, 2026 Maxim [maxirmx] Samsonov (www.sw.consulting) +// Copyright (C) 2025, 2026 Maxim [maxirmx] Samsonov (www.sw.consulting) // All rights reserved. // This file is a part of fuelflux application @@ -55,10 +55,13 @@ Controller::Controller(ControllerId controllerId, Controller::Controller(ControllerId controllerId, std::shared_ptr backend, std::chrono::seconds noFlowCancelTimeout, - ControllerPersistencePaths persistencePaths) + ControllerPersistencePaths persistencePaths, + std::chrono::milliseconds foregroundWait) : controllerId_(std::move(controllerId)) , stateMachine_(this) , backend_(backend ? std::move(backend) : CreateDefaultBackend()) + , backendPrototype_(backend_) + , foregroundWait_(foregroundWait) , selectedTank_(0) , enteredVolume_(0.0) , selectedIntakeDirection_(IntakeDirection::In) @@ -67,6 +70,9 @@ Controller::Controller(ControllerId controllerId, , isRunning_(false) , noFlowCancelTimeout_(noFlowCancelTimeout) { + if (foregroundWait_.count() <= 0) throw std::invalid_argument("Foreground wait must be positive"); + if (!backendPrototype_->CreateIndependentSession()) + throw std::invalid_argument("Controller requires a backend with independent cancellable sessions"); resetSessionData(); // Initialize user cache and cache manager @@ -75,7 +81,9 @@ Controller::Controller(ControllerId controllerId, // Create a separate backend instance for cache manager synchronization to avoid JWT token conflicts // The cache manager needs its own backend with independent session state so that synchronization // operations don't interfere with concurrent user authorization sessions in the main backend - auto syncBackend = CreateDefaultBackendShared(backend_->GetControllerUid(), nullptr); + auto syncBackend = backendPrototype_->CreateIndependentSession(); + if (!syncBackend) + throw std::invalid_argument("Controller requires a backend with independent cancellable sessions"); cacheManager_ = std::make_shared(userCache_, syncBackend); LOG_CTRL_INFO("User cache initialized at: {}", persistencePaths.cacheDbPath); } catch (const std::exception& e) { @@ -85,6 +93,8 @@ Controller::Controller(ControllerId controllerId, try { messageStorage_ = std::make_shared(persistencePaths.messageStorageDbPath); + reportWorker_ = std::make_unique(messageStorage_, backendPrototype_, timing::kBacklogWorkerInterval); + if (cacheManager_) cacheManager_->SetReportStorage(messageStorage_); LOG_CTRL_INFO("Message storage initialized at: {}", persistencePaths.messageStorageDbPath); const auto storedCoefficient = messageStorage_->GetCalibrationCoefficient(); if (storedCoefficient.has_value()) { @@ -100,20 +110,24 @@ Controller::Controller(ControllerId controllerId, } Controller::~Controller() { - shutdown(); + // Destruction with a live loop would free state still used by that thread. + if (!shutdown()) std::terminate(); } bool Controller::initialize() { LOG_CTRL_INFO("Initializing controller: {}", controllerId_); lastErrorMessage_.clear(); + peripheralsNeedShutdown_ = false; bool ok = initializePeripherals(); + peripheralsNeedShutdown_ = ok; // Setup peripheral callbacks setupPeripheralCallbacks(); // Initialize state machine stateMachine_.initialize(); + if (reportWorker_) reportWorker_->Start(); // Start cache manager (non-blocking) if (cacheManager_) { @@ -138,39 +152,51 @@ bool Controller::initialize() { return ok; } -void Controller::shutdown() { +bool Controller::shutdown() { + std::lock_guard shutdownLock(shutdownMutex_); LOG_CTRL_INFO("Shutting down..."); + // Network jobs hold their own state and never reference Controller. + { + std::lock_guard lock(lifecycleMutex_); + shutdownRequested_ = true; + isRunning_ = false; + } + eventCv_.notify_all(); // Stop cache manager first if (cacheManager_) { cacheManager_->Stop(); LOG_CTRL_INFO("Cache manager stopped"); } - if (isRunning_) { - isRunning_ = false; - stopNoFlowMonitorThread(); - eventCv_.notify_all(); - - // Wait for the event loop thread to actually exit - // The thread checks isRunning_ at the top of the loop and the - // condition variable wait has a kEventLoopWaitInterval timeout, so this waits up to kShutdownDeadline - const auto deadline = std::chrono::steady_clock::now() + timing::kShutdownDeadline; - while (!threadExited_ && std::chrono::steady_clock::now() < deadline) { - std::this_thread::sleep_for(timing::kEventLoopIdleSleep); - } - - if (!threadExited_) { - LOG_CTRL_ERROR("Thread shutdown timeout - thread did not exit within 2 seconds"); + stopNoFlowMonitorThread(); + // Do not tear down state while a controller-side SQLite transaction or + // peripheral action is still finishing. The caller joins the loop thread. + { + std::unique_lock lock(lifecycleMutex_); + if (!lifecycleCv_.wait_for(lock, timing::kShutdownDeadline, [this] { return !loopActive_; })) { + LOG_CTRL_ERROR("Controller shutdown deadline exceeded; live state has not been torn down"); + return false; } - - // Shutdown peripherals + } + abandonAuthorization(); + authorizationExecutor_.Shutdown(); + if (reportWorker_) reportWorker_->Stop(); + if (!sessionAuthorizedFromCache_ && backend_ && backend_->IsAuthorized()) { + try { (void)backend_->Deauthorize(); } catch (...) {} + } + backend_.reset(); + if (peripheralsNeedShutdown_) { shutdownPeripherals(); + peripheralsNeedShutdown_ = false; } LOG_CTRL_INFO("Shutdown complete"); + return true; } bool Controller::reinitializeDevice() { + abandonAuthorization(); + foregroundReport_.reset(); LOG_CTRL_WARN("Reinitializing device after error"); lastErrorMessage_.clear(); @@ -178,15 +204,17 @@ bool Controller::reinitializeDevice() { // but do NOT stop the event loop - we need it to process the ErrorRecovery event { std::lock_guard lock(eventQueueMutex_); - std::queue emptyQueue; + std::queue emptyQueue; std::swap(eventQueue_, emptyQueue); } // Shutdown old peripherals shutdownPeripherals(); + peripheralsNeedShutdown_ = false; // Reinitialize peripherals and callbacks bool ok = initializePeripherals(); + peripheralsNeedShutdown_ = ok; if (ok) { setupPeripheralCallbacks(); } @@ -204,14 +232,23 @@ bool Controller::reinitializeDevice() { } void Controller::run() { + { + std::lock_guard lock(lifecycleMutex_); + if (shutdownRequested_ || !isRunning_) return; + if (loopActive_) throw std::logic_error("Controller event loop already running"); + loopActive_ = true; + } + struct ExitGuard { + Controller* controller; + ~ExitGuard() { controller->finishRun(); } + } exitGuard{this}; LOG_CTRL_INFO("Starting main loop"); - // Reset the flag at the start of the run loop - threadExited_ = false; - while (isRunning_) { + pollBackendOperations(); bool haveEvent = false; Event event = Event::Timeout; // initialize but treat as invalid until popped + QueuedEvent queued{Event::Timeout}; { std::unique_lock lock(eventQueueMutex_); if (eventQueue_.empty()) { @@ -219,20 +256,32 @@ void Controller::run() { eventCv_.wait_for(lock, timing::kEventLoopWaitInterval, [this] { return !eventQueue_.empty() || !isRunning_; }); } if (!eventQueue_.empty()) { - event = eventQueue_.front(); + queued = eventQueue_.front(); + event = queued.event; eventQueue_.pop(); haveEvent = true; } } + if (!isRunning_) break; if (haveEvent) { + if (queued.authorizationCancel) { + if (queued.cancelEnabled && queued.authorizationId == activeAuthorizationId_.load() && + stateMachine_.getCurrentState() == SystemState::Authorization) + cancelSlowAuthorization(); + continue; + } // Handle DisplayReset event in the controller thread to avoid race conditions. // This event bypasses the state machine because display reset is a hardware // operation that doesn't affect logical state transitions. The state machine // state is preserved across display resets, and the display simply shows the // same state information after reinitialization. This design keeps display // hardware management separate from business logic. - if (event == Event::DisplayReset) { + if (event == Event::FlowDisplayRefresh) { + // A meter refresh queued before report completion must never + // be interpreted as PIN input on the completion screen. + updateDisplay(); + } else if (event == Event::DisplayReset) { reinitializeDisplay(); } else { stateMachine_.processEvent(event); @@ -243,16 +292,29 @@ void Controller::run() { } } - // Signal that thread has exited the main loop - threadExited_ = true; LOG_CTRL_INFO("Main loop stopped"); } +void Controller::finishRun() { + { + std::lock_guard lock(lifecycleMutex_); + isRunning_ = false; + loopActive_ = false; + } + lifecycleCv_.notify_all(); +} + // Allow other threads to post events into controller's loop void Controller::postEvent(Event event) { + QueuedEvent queued{event}; + if (event == Event::CancelPressed && stateMachine_.getCurrentState() == SystemState::Authorization) { + queued.authorizationCancel = true; + queued.authorizationId = activeAuthorizationId_.load(); + queued.cancelEnabled = authorizationSlow_.load(); + } { std::lock_guard lock(eventQueueMutex_); - eventQueue_.push(event); + eventQueue_.push(queued); } eventCv_.notify_one(); } @@ -261,7 +323,7 @@ void Controller::postEvent(Event event) { // leaving the first non-InputUpdated event (if any) untouched. void Controller::discardPendingInputUpdatedEvents() { std::lock_guard lock(eventQueueMutex_); - while (!eventQueue_.empty() && eventQueue_.front() == Event::InputUpdated) { + while (!eventQueue_.empty() && eventQueue_.front().event == Event::InputUpdated) { eventQueue_.pop(); } } @@ -451,7 +513,7 @@ void Controller::handleFlowUpdate(Volume currentVolume) { auto now = std::chrono::steady_clock::now(); if ((now - lastFlowCallbackTime_) >= timing::kFlowDisplayRefreshInterval) { lastFlowCallbackTime_ = now; - postEvent(Event::InputUpdated); + postEvent(Event::FlowDisplayRefresh); } } @@ -515,6 +577,9 @@ void Controller::startNewSession() { } void Controller::endCurrentSession() { + abandonAuthorization(); + if (!sessionAuthorizedFromCache_ && backend_ && backend_->IsAuthorized()) + (void)backend_->Deauthorize(); resetSessionData(); clearInputSilent(); if (pump_ && pump_->isRunning()) { @@ -523,9 +588,6 @@ void Controller::endCurrentSession() { if (flowMeter_) { flowMeter_->stopMeasurement(); } - if (!sessionAuthorizedFromCache_ && backend_ && backend_->IsAuthorized()) { - (void)backend_->Deauthorize(); - } } void Controller::clearInput() { @@ -600,7 +662,174 @@ void Controller::setMaxValue() { } // Authorization +Controller::AuthorizationAttempt::~AuthorizationAttempt() { + // The last owner can be either the worker or controller. An adopted session + // belongs to the controller; every other successful session is discarded. + if (done.load() && !adopted && backend && backend->IsAuthorized()) { + try { backend->Deauthorize(); } catch (...) {} + } +} + +SavedAuthorizationState Controller::savedAuthorization(const std::string& uid) const { + if (!messageStorage_) return {}; + return messageStorage_->CaptureAuthorizationState(uid, [this, &uid] { + return userCache_ ? userCache_->GetAuthorizationSnapshot(uid) : std::nullopt; + }); +} + +void Controller::applyAuthorization(const AuthorizationSnapshot& snapshot, bool fromCache) { + sessionAuthorizedFromCache_ = fromCache; + currentUser_ = snapshot.user; + if (fromCache) currentUser_.price = 0.0; + cachedFuelTanks_ = snapshot.tanks; + availableTanks_.clear(); + for (const auto& tank : snapshot.tanks) { + TankInfo info; + info.number = tank.visualNumberTank; + availableTanks_.push_back(info); + } + if (!fromCache) { + // Keep protection until a synchronization fetch begun after resolution + // commits. Otherwise an older in-flight sync could undo this fresh reply. + if (messageStorage_) messageStorage_->RefreshResolvedSnapshot(snapshot); + if (cacheManager_) cacheManager_->UpdateCacheEntry(currentUser_.uid, currentUser_.allowance, + static_cast(currentUser_.role)); + } +} + +void Controller::abandonAuthorization() { + activeAuthorizationId_.store(0); + if (authorizationAttempt_) { + authorizationAttempt_->backend->CancelPendingRequests(); + authorizationAttempt_.reset(); + } + authorizationSlow_ = false; +} + +void Controller::beginAuthorization(const UserId& uid) { + abandonAuthorization(); + if (!messageStorage_) { + showError("Ошибка записи"); + postEvent(Event::AuthorizationFailed); + return; + } + const auto started = std::chrono::steady_clock::now(); + const auto local = savedAuthorization(uid); + if (!local.reportStorageAvailable) { + showError("Ошибка записи"); + postEvent(Event::AuthorizationFailed); + return; + } + const auto saved = local.reportStorageAvailable ? local.saved : std::nullopt; + if (messageStorage_ && local.pendingReports) { + if (saved) { + applyAuthorization(*saved, true); + postEvent(Event::AuthorizationSuccess); + } else postEvent(Event::AuthorizationFailed); + return; + } + auto session = backendPrototype_->CreateIndependentSession(); + if (!session) { postEvent(Event::AuthorizationFailed); return; } + auto attempt = std::make_shared(); + attempt->id = ++nextAuthorizationId_; + activeAuthorizationId_.store(attempt->id); + attempt->uid = uid; + attempt->saved = saved; + attempt->started = started; + attempt->backend = std::move(session); + authorizationAttempt_ = attempt; + if (!authorizationExecutor_.Submit([attempt]() { + try { + attempt->success = attempt->backend->Authorize(attempt->uid); + attempt->networkError = attempt->backend->IsNetworkError(); + if (attempt->success) { + attempt->online.user = {attempt->uid, static_cast(attempt->backend->GetRoleId()), + attempt->backend->GetAllowance(), attempt->backend->GetPrice()}; + attempt->online.tanks = attempt->backend->GetFuelTanks(); + } + } catch (...) { attempt->success = false; attempt->networkError = true; } + attempt->done.store(true); + })) { + abandonAuthorization(); + if (saved) { applyAuthorization(*saved, true); postEvent(Event::AuthorizationSuccess); } + else postEvent(Event::AuthorizationFailed); + } +} + +void Controller::cancelSlowAuthorization() { + if (!authorizationSlow_ || !authorizationAttempt_) return; + abandonAuthorization(); + stateMachine_.processEvent(Event::AuthorizationCancelled); +} + +void Controller::pollBackendOperations() { + if (authorizationAttempt_ && stateMachine_.getCurrentState() != SystemState::Authorization) + abandonAuthorization(); + if (authorizationAttempt_) { + auto attempt = authorizationAttempt_; + if (attempt->done.load()) { + authorizationAttempt_.reset(); + activeAuthorizationId_.store(0); + authorizationSlow_ = false; + if (attempt->success) { + if (attempt->online.tanks.empty()) { + stateMachine_.processEvent(Event::AuthorizationDenied); + return; + } + attempt->adopted = true; + backend_ = attempt->backend; + applyAuthorization(attempt->online, false); + stateMachine_.processEvent(Event::AuthorizationSuccess); + } else if (attempt->networkError && attempt->saved) { + applyAuthorization(*attempt->saved, true); + stateMachine_.processEvent(Event::AuthorizationSuccess); + } else stateMachine_.processEvent(attempt->networkError ? Event::AuthorizationFailed : Event::AuthorizationDenied); + } else if (!authorizationSlow_ && std::chrono::steady_clock::now() - attempt->started >= foregroundWait_) { + if (attempt->saved) { + const auto saved = *attempt->saved; + abandonAuthorization(); + applyAuthorization(saved, true); + stateMachine_.processEvent(Event::AuthorizationSuccess); + } else { + authorizationSlow_ = true; + updateDisplay(); + } + } + } + if (foregroundReport_ && (!messageStorage_->HasReport(*foregroundReport_) || + std::chrono::steady_clock::now() - reportStarted_ >= foregroundWait_)) { + foregroundReport_.reset(); + const auto state = stateMachine_.getCurrentState(); + if (state == SystemState::RefuelDataTransmission || state == SystemState::IntakeDataTransmission) + stateMachine_.processEvent(Event::DataTransmissionComplete); + } +} + void Controller::requestAuthorization(const UserId& userId) { + if (!messageStorage_) { + showError("Ошибка записи"); + postEvent(Event::AuthorizationFailed); + return; + } + const auto local = savedAuthorization(userId); + if (!local.reportStorageAvailable) { + showError("Ошибка записи"); + postEvent(Event::AuthorizationFailed); + return; + } + const auto saved = local.saved; + if (local.pendingReports) { + if (saved) { + applyAuthorization(*saved, true); + postEvent(Event::AuthorizationSuccess); + } else { + postEvent(Event::AuthorizationFailed); + } + return; + } + if (!backend_) { + backend_ = backendPrototype_->CreateIndependentSession(); + } if (!backend_) { showError("Backend unavailable"); postEvent(Event::AuthorizationFailed); @@ -609,61 +838,25 @@ void Controller::requestAuthorization(const UserId& userId) { // This method handles the actual authorization for both card and PIN if (backend_->Authorize(userId)) { - sessionAuthorizedFromCache_ = false; - currentUser_.uid = userId; - currentUser_.role = static_cast(backend_->GetRoleId()); - currentUser_.allowance = backend_->GetAllowance(); - currentUser_.price = backend_->GetPrice(); - - availableTanks_.clear(); - cachedFuelTanks_.clear(); - for (const auto& tank : backend_->GetFuelTanks()) { - TankInfo info; - info.number = tank.visualNumberTank; - availableTanks_.push_back(info); - cachedFuelTanks_.push_back(tank); - } - - // Update cache with authorization data - if (cacheManager_) { - cacheManager_->UpdateCacheEntry(userId, currentUser_.allowance, - static_cast(currentUser_.role)); + AuthorizationSnapshot online{{userId, static_cast(backend_->GetRoleId()), + backend_->GetAllowance(), backend_->GetPrice()}, + backend_->GetFuelTanks()}; + if (online.tanks.empty()) { + if (backend_->IsAuthorized()) (void)backend_->Deauthorize(); + postEvent(Event::AuthorizationDenied); + return; } - - // Post event instead of processing it directly to maintain sequential event processing + applyAuthorization(online, false); postEvent(Event::AuthorizationSuccess); } else { // Check if it's a network error bool isNetworkError = backend_->IsNetworkError(); - // Try cache fallback if network error and cache is available - if (isNetworkError && userCache_ && messageStorage_) { - auto cached = userCache_->GetEntry(userId); - if (cached.has_value()) { - sessionAuthorizedFromCache_ = true; - currentUser_.uid = cached->uid; - currentUser_.role = static_cast(cached->roleId); - currentUser_.allowance = cached->allowance; - currentUser_.price = 0.0; - availableTanks_.clear(); - cachedFuelTanks_.clear(); - const auto cachedTanks = userCache_->GetTanks(); - for (const auto& tank : cachedTanks) { - TankInfo info; - info.number = tank.visualNumberTank; - availableTanks_.push_back(info); - - BackendTankInfo cachedInfo; - cachedInfo.idTank = tank.idTank; - cachedInfo.visualNumberTank = tank.visualNumberTank; - cachedInfo.nameTank = tank.nameTank; - cachedInfo.volume = tank.volume; - cachedFuelTanks_.push_back(cachedInfo); - } - LOG_CTRL_WARN("Authorized user {} from cache due to backend network error", userId); - postEvent(Event::AuthorizationSuccess); - return; - } + if (isNetworkError && saved) { + applyAuthorization(*saved, true); + LOG_CTRL_WARN("Authorized user {} from cache due to backend network error", userId); + postEvent(Event::AuthorizationSuccess); + return; } // Post appropriate failure event @@ -829,61 +1022,52 @@ void Controller::completeIntakeOperation() { } // Transaction logging -void Controller::logRefuelTransaction(const RefuelTransaction& transaction) { - if (sessionAuthorizedFromCache_ && messageStorage_) { - const auto timestampMs = std::chrono::duration_cast( - transaction.timestamp.time_since_epoch()).count(); - - nlohmann::json payload; - payload["TankNumber"] = transaction.tankNumber; - payload["FuelVolume"] = transaction.volume; - payload["TimeAt"] = timestampMs; - - const bool stored = messageStorage_->AddBacklog(transaction.userId, MessageMethod::Refuel, payload.dump()); - if (!stored) { - LOG_CTRL_ERROR("Failed to save offline refuel report to backlog for user {}", transaction.userId); - } - - if (cacheManager_ && currentUser_.role == UserRole::Customer) { - cacheManager_->DeductAllowance(transaction.userId, transaction.volume); - } - return; +bool Controller::submitReport(MessageMethod method, const std::string& uid, TankNumber tankNumber, + Volume volume, std::chrono::system_clock::time_point timestamp, IntakeDirection direction) { + reportStarted_ = std::chrono::steady_clock::now(); + const auto tank = std::find_if(cachedFuelTanks_.begin(), cachedFuelTanks_.end(), + [tankNumber](const BackendTankInfo& t) { return t.visualNumberTank == tankNumber; }); + if (!reportWorker_ || !messageStorage_ || tank == cachedFuelTanks_.end()) { + showError("Ошибка записи"); + postEvent(Event::Error); + return false; } - - if (backend_) { - (void)backend_->Refuel(transaction.tankNumber, transaction.volume); - - // Deduct allowance from cache for customers (RoleId==1) - // Do this even if refuel fails, but check we're not processing backlog - if (cacheManager_ && currentUser_.role == UserRole::Customer) { - cacheManager_->DeductAllowance(transaction.userId, transaction.volume); + nlohmann::json payload{{"TankNumber", tank->idTank}, + {"TimeAt", std::chrono::duration_cast(timestamp.time_since_epoch()).count()}}; + if (method == MessageMethod::Refuel) payload["FuelVolume"] = volume; + else { payload["IntakeVolume"] = volume; payload["Direction"] = static_cast(direction); } + AuthorizationSnapshot snapshot{currentUser_, cachedFuelTanks_}; + snapshot.user.uid = uid; + auto session = !sessionAuthorizedFromCache_ ? backend_ : nullptr; + const auto id = reportWorker_->Submit(method, payload.dump(), snapshot, + method == MessageMethod::Refuel ? volume : 0.0, session); + if (!id) { + if (session) { + try { session->Deauthorize(); } catch (...) {} + if (session == backend_) { + backend_.reset(); + } } + showError("Ошибка записи"); + postEvent(Event::Error); + return false; } + if (session) backend_.reset(); // The reporting worker now owns this session. + if (method == MessageMethod::Refuel && cacheManager_) + cacheManager_->DeductAllowance(uid, volume); // Mirror; protected durable state is authoritative. + foregroundReport_ = *id; + return true; } -void Controller::logIntakeTransaction(const IntakeTransaction& transaction) { - if (sessionAuthorizedFromCache_ && messageStorage_) { - const auto timestampMs = std::chrono::duration_cast( - transaction.timestamp.time_since_epoch()).count(); - - nlohmann::json payload; - payload["TankNumber"] = transaction.tankNumber; - payload["IntakeVolume"] = transaction.volume; - payload["Direction"] = static_cast(transaction.direction); - payload["TimeAt"] = timestampMs; - - const bool stored = messageStorage_->AddBacklog(transaction.operatorId, MessageMethod::Intake, payload.dump()); - if (!stored) { - LOG_CTRL_ERROR("Failed to save offline intake report to backlog for user {}", transaction.operatorId); - } - return; - } - - if (backend_) { - (void)backend_->Intake(transaction.tankNumber, transaction.volume, transaction.direction); - } +void Controller::logRefuelTransaction(const RefuelTransaction& transaction) { + submitReport(MessageMethod::Refuel, transaction.userId, transaction.tankNumber, + transaction.volume, transaction.timestamp); } +void Controller::logIntakeTransaction(const IntakeTransaction& transaction) { + submitReport(MessageMethod::Intake, transaction.operatorId, transaction.tankNumber, + transaction.volume, transaction.timestamp, transaction.direction); +} // Utility functions std::string Controller::formatVolume(Volume volume) const { std::ostringstream oss; diff --git a/src/main.cpp b/src/main.cpp index 635e85d..fd49e77 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -1,4 +1,4 @@ -// Copyright (C) 2025, 2026 Maxim [maxirmx] Samsonov (www.sw.consulting) +// Copyright (C) 2025, 2026 Maxim [maxirmx] Samsonov (www.sw.consulting) // All rights reserved. // This file is a part of fuelflux application @@ -39,6 +39,7 @@ #include #include #include +#include #include #include #include @@ -198,11 +199,7 @@ int main([[maybe_unused]] int argc, [[maybe_unused]] char* argv[]) { msg.line3 = "Очередь"; display->showMessage(msg); - auto storage = std::make_shared(STORAGE_DB_PATH); - auto backend = Controller::CreateDefaultBackend(storage); - auto backlogBackend = Controller::CreateDefaultBackendShared(controllerId, nullptr); - BacklogWorker backlogWorker(storage, backlogBackend, timing::kBacklogWorkerInterval); - backlogWorker.Start(); + auto backend = Controller::CreateDefaultBackend(); // ----- Контроллер ----- @@ -298,26 +295,40 @@ int main([[maybe_unused]] int argc, [[maybe_unused]] char* argv[]) { // Start controller main loop in a separate thread controller.updateDisplay(); - std::thread controllerThread([&controller]() { - controller.run(); + std::atomic controllerFinished{false}; + std::exception_ptr controllerFailure; + std::thread controllerThread([&]() { + try { controller.run(); } + catch (...) { controllerFailure = std::current_exception(); } + controllerFinished = true; }); + struct JoinLoop { + std::thread& thread; + ~JoinLoop() { if (thread.joinable()) thread.join(); } + } joinLoop{controllerThread}; + const auto shutdownController = [&controller] { + if (!controller.shutdown()) { + // A stuck peripheral action cannot be safely detached from + // Controller. Let the service supervisor restart the process. + std::cerr << "Controller shutdown timed out; terminating for restart" << std::endl; + std::_Exit(EXIT_FAILURE); + } + }; // Ensure threads are cleaned up even if exceptions occur try { // Main thread waits for shutdown signal - while (g_running) { + while (g_running && !controllerFinished) { std::this_thread::sleep_for(timing::kMainLoopWaitInterval); } LOG_INFO("Shutting down..."); // Shutdown controller - controller.shutdown(); - backlogWorker.Stop(); + shutdownController(); } catch (...) { // Ensure controller is stopped even if exception occurs - controller.shutdown(); - backlogWorker.Stop(); + shutdownController(); throw; // Re-throw to be caught by outer handler } @@ -327,6 +338,7 @@ int main([[maybe_unused]] int argc, [[maybe_unused]] char* argv[]) { if (controllerThread.joinable()) { controllerThread.join(); } + if (controllerFailure) std::rethrow_exception(controllerFailure); LOG_INFO("Shutdown complete"); } catch (const std::exception& e) { @@ -342,7 +354,7 @@ int main([[maybe_unused]] int argc, [[maybe_unused]] char* argv[]) { if (g_running) { if (retryCount >= MAX_RETRIES) { LOG_CRITICAL("Maximum retry limit ({}) reached, entering permanent failure state", MAX_RETRIES); - + BackendBase::ShutdownAsyncRequests(); #ifdef USE_CARES if (caresInitialized) { @@ -391,6 +403,7 @@ int main([[maybe_unused]] int argc, [[maybe_unused]] char* argv[]) { } } + BackendBase::ShutdownAsyncRequests(); #ifdef USE_CARES if (caresInitialized) { CleanupCaresLibrary(); diff --git a/src/message_storage.cpp b/src/message_storage.cpp index 4d996ad..a0f378a 100644 --- a/src/message_storage.cpp +++ b/src/message_storage.cpp @@ -8,9 +8,62 @@ #include #include #include +#include namespace fuelflux { +namespace { +class Statement { +public: + Statement(sqlite3* db, const char* sql) { + if (sqlite3_prepare_v2(db, sql, -1, &value, nullptr) != SQLITE_OK) + throw std::runtime_error(sqlite3_errmsg(db)); + } + ~Statement() { sqlite3_finalize(value); } + void text(int n, const std::string& s) { sqlite3_bind_text(value, n, s.c_str(), -1, SQLITE_TRANSIENT); } + void number(int n, long long v) { sqlite3_bind_int64(value, n, v); } + sqlite3_stmt* value = nullptr; +}; + +class Transaction { +public: + explicit Transaction(sqlite3* db) : db_(db) { + if (sqlite3_exec(db_, "BEGIN IMMEDIATE", nullptr, nullptr, nullptr) != SQLITE_OK) + throw std::runtime_error(sqlite3_errmsg(db_)); + } + ~Transaction() { if (!done_) sqlite3_exec(db_, "ROLLBACK", nullptr, nullptr, nullptr); } + bool commit() { done_ = sqlite3_exec(db_, "COMMIT", nullptr, nullptr, nullptr) == SQLITE_OK; return done_; } +private: + sqlite3* db_; + bool done_ = false; +}; + +std::string columnText(sqlite3_stmt* stmt, int n) { + const auto* p = sqlite3_column_text(stmt, n); + return p ? reinterpret_cast(p) : ""; +} + +nlohmann::json snapshotJson(const AuthorizationSnapshot& s) { + nlohmann::json tanks = nlohmann::json::array(); + for (const auto& t : s.tanks) + tanks.push_back({{"id", t.idTank}, {"visual", t.visualNumberTank}, {"name", t.nameTank}, {"volume", t.volume}}); + return {{"uid", s.user.uid}, {"role", static_cast(s.user.role)}, + {"allowance", s.user.allowance}, {"tanks", tanks}}; +} + +AuthorizationSnapshot parseSnapshot(const std::string& text) { + auto j = nlohmann::json::parse(text); + AuthorizationSnapshot s; + s.user.uid = j.at("uid").get(); + s.user.role = static_cast(j.at("role").get()); + s.user.allowance = j.at("allowance").get(); + for (const auto& t : j.at("tanks")) + s.tanks.push_back({t.at("id").get(), t.at("visual").get(), + t.at("name").get(), t.at("volume").get()}); + return s; +} +} // namespace + MessageStorage::MessageStorage(const std::string& dbPath) : db_(nullptr) , dbPath_(dbPath) { @@ -30,10 +83,41 @@ if (sqlite3_open(dbPath.c_str(), &db) != SQLITE_OK) { db_ = db; sqlite3_busy_timeout(db, 5000); - Execute("CREATE TABLE IF NOT EXISTS backlog (uid TEXT NOT NULL, method TEXT NOT NULL, data TEXT NOT NULL);"); - Execute("CREATE TABLE IF NOT EXISTS dead_messages (uid TEXT NOT NULL, method TEXT NOT NULL, data TEXT NOT NULL);"); - Execute("CREATE TABLE IF NOT EXISTS device_settings (key TEXT PRIMARY KEY, value REAL NOT NULL);"); - Execute("INSERT OR IGNORE INTO device_settings (key, value) VALUES ('calibration_coefficient', 1.0);"); + // Migrate legacy implicit rowids, preserving their order and interpretation. + try { + Transaction tx(db_); + bool legacy = false; + { + Statement exists(db_, "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='backlog'"); + sqlite3_step(exists.value); + if (sqlite3_column_int(exists.value, 0)) { + Statement columns(db_, "PRAGMA table_info(backlog)"); + legacy = true; + while (sqlite3_step(columns.value) == SQLITE_ROW) + if (columnText(columns.value, 1) == "attempt") legacy = false; + } + } + if (legacy && sqlite3_exec(db_, "ALTER TABLE backlog RENAME TO backlog_legacy", nullptr, nullptr, nullptr) != SQLITE_OK) + throw std::runtime_error(sqlite3_errmsg(db_)); + if (sqlite3_exec(db_, "CREATE TABLE IF NOT EXISTS backlog (id INTEGER PRIMARY KEY AUTOINCREMENT, uid TEXT NOT NULL, method TEXT NOT NULL, data TEXT NOT NULL, attempt INTEGER NOT NULL DEFAULT 0, in_flight INTEGER NOT NULL DEFAULT 0, canonical INTEGER NOT NULL DEFAULT 0, retry_after INTEGER NOT NULL DEFAULT 0)", nullptr, nullptr, nullptr) != SQLITE_OK) + throw std::runtime_error(sqlite3_errmsg(db_)); + if (legacy && sqlite3_exec(db_, "INSERT INTO backlog(id,uid,method,data) SELECT rowid,uid,method,data FROM backlog_legacy ORDER BY rowid; DROP TABLE backlog_legacy", nullptr, nullptr, nullptr) != SQLITE_OK) + throw std::runtime_error(sqlite3_errmsg(db_)); + if (sqlite3_exec(db_, "CREATE TABLE IF NOT EXISTS card_report_state (uid TEXT PRIMARY KEY, snapshot TEXT NOT NULL, revision INTEGER NOT NULL DEFAULT 1, protected INTEGER NOT NULL DEFAULT 1)", nullptr, nullptr, nullptr) != SQLITE_OK || !tx.commit()) + throw std::runtime_error(sqlite3_errmsg(db_)); + } catch (...) { + sqlite3_close(db_); + db_ = nullptr; + throw; + } + if (!Execute("CREATE TABLE IF NOT EXISTS dead_messages (uid TEXT NOT NULL, method TEXT NOT NULL, data TEXT NOT NULL);") || + !Execute("CREATE TABLE IF NOT EXISTS device_settings (key TEXT PRIMARY KEY, value REAL NOT NULL);") || + !Execute("INSERT OR IGNORE INTO device_settings (key, value) VALUES ('calibration_coefficient', 1.0);")) { + const std::string error = sqlite3_errmsg(db_); + sqlite3_close(db_); + db_ = nullptr; + throw std::runtime_error("Failed to initialize report storage: " + error); + } } MessageStorage::~MessageStorage() { @@ -49,6 +133,34 @@ bool MessageStorage::IsOpen() const { return db_ != nullptr; } +SavedAuthorizationState MessageStorage::CaptureAuthorizationState(const std::string& uid, + const std::function()>& generalCache) const { + std::lock_guard lock(dbMutex_); + SavedAuthorizationState result; + if (!db_) return result; + try { + Transaction tx(db_); + Statement pending(db_, "SELECT EXISTS(SELECT 1 FROM backlog WHERE uid=?)"); + pending.text(1, uid); + if (sqlite3_step(pending.value) != SQLITE_ROW) return result; + result.pendingReports = sqlite3_column_int(pending.value, 0) != 0; + // Probe writable storage; the transaction is rolled back on return. + // A later write can still fail (for example, the disk fills during fueling). + Statement writable(db_, "UPDATE device_settings SET value=value WHERE key='calibration_coefficient'"); + if (sqlite3_step(writable.value) != SQLITE_DONE) return result; + Statement card(db_, "SELECT snapshot FROM card_report_state WHERE uid=? AND protected=1"); + card.text(1, uid); + const int status = sqlite3_step(card.value); + if (status == SQLITE_ROW) result.saved = parseSnapshot(columnText(card.value, 0)); + else if (status == SQLITE_DONE && !result.pendingReports && generalCache) { + result.saved = generalCache(); + } + else if (status != SQLITE_DONE) return result; + result.reportStorageAvailable = true; + } catch (...) { result.saved.reset(); } + return result; +} + bool MessageStorage::Execute(const std::string& sql) const { std::lock_guard lock(dbMutex_); if (!db_) { @@ -134,7 +246,7 @@ std::optional MessageStorage::GetNextBacklog() { } sqlite3_stmt* stmt = nullptr; - const char* sql = "SELECT rowid, uid, method, data FROM backlog ORDER BY rowid ASC LIMIT 1;"; + const char* sql = "SELECT id, uid, method, data, attempt, canonical FROM backlog ORDER BY id ASC LIMIT 1;"; if (sqlite3_prepare_v2(db_, sql, -1, &stmt, nullptr) != SQLITE_OK) { return std::nullopt; } @@ -160,6 +272,8 @@ std::optional MessageStorage::GetNextBacklog() { return std::nullopt; } message.method = *method; + message.attempt = sqlite3_column_int64(stmt, 4); + message.canonicalTankId = sqlite3_column_int(stmt, 5) != 0; if (data) { message.data = data; } @@ -189,6 +303,143 @@ bool MessageStorage::RemoveBacklog(long long id) { return ok; } +std::optional MessageStorage::EnqueueReport(MessageMethod method, const std::string& data, + AuthorizationSnapshot snapshot, double deduction) { + std::lock_guard lock(dbMutex_); + if (!db_ || snapshot.user.uid.empty() || snapshot.tanks.empty() || !std::isfinite(deduction) || deduction < 0) + return std::nullopt; + try { + Transaction tx(db_); + snapshot.user.allowance = std::max(0.0, snapshot.user.allowance - deduction); + Statement card(db_, "INSERT INTO card_report_state(uid,snapshot) VALUES(?,?) ON CONFLICT(uid) DO UPDATE SET snapshot=excluded.snapshot, revision=revision+1, protected=1"); + card.text(1, snapshot.user.uid); card.text(2, snapshotJson(snapshot).dump()); + if (sqlite3_step(card.value) != SQLITE_DONE) return std::nullopt; + Statement report(db_, "INSERT INTO backlog(uid,method,data,canonical) VALUES(?,?,?,1)"); + report.text(1, snapshot.user.uid); report.text(2, MethodToString(method)); report.text(3, data); + if (sqlite3_step(report.value) != SQLITE_DONE) return std::nullopt; + const auto id = sqlite3_last_insert_rowid(db_); + if (!tx.commit()) return std::nullopt; + return id; + } catch (...) { return std::nullopt; } +} + +std::optional MessageStorage::GetProtectedSnapshot(const std::string& uid) const { + std::lock_guard lock(dbMutex_); + if (!db_) return std::nullopt; + try { + Statement q(db_, "SELECT snapshot, EXISTS(SELECT 1 FROM backlog WHERE uid=?) FROM card_report_state WHERE uid=? AND protected=1"); + q.text(1, uid); q.text(2, uid); + if (sqlite3_step(q.value) != SQLITE_ROW) return std::nullopt; + return ProtectedCardSnapshot{parseSnapshot(columnText(q.value, 0)), sqlite3_column_int(q.value, 1) != 0}; + } catch (...) { return std::nullopt; } +} + +bool MessageStorage::HasPendingReports(const std::string& uid) const { + std::lock_guard lock(dbMutex_); + try { + Statement q(db_, "SELECT EXISTS(SELECT 1 FROM backlog WHERE uid=?)"); q.text(1, uid); + return sqlite3_step(q.value) != SQLITE_ROW || sqlite3_column_int(q.value, 0) != 0; + } catch (...) { return true; } +} + +bool MessageStorage::HasReport(long long id) const { + std::lock_guard lock(dbMutex_); + try { + Statement q(db_, "SELECT EXISTS(SELECT 1 FROM backlog WHERE id=?)"); q.number(1, id); + return sqlite3_step(q.value) != SQLITE_ROW || sqlite3_column_int(q.value, 0) != 0; + } catch (...) { return true; } +} + +bool MessageStorage::ClearProtectedSnapshot(const std::string& uid) { + std::lock_guard lock(dbMutex_); + try { + Statement q(db_, "UPDATE card_report_state SET protected=0,revision=revision+1 WHERE uid=? AND protected=1 AND NOT EXISTS(SELECT 1 FROM backlog WHERE uid=?)"); + q.text(1, uid); q.text(2, uid); + return sqlite3_step(q.value) == SQLITE_DONE; + } catch (...) { return false; } +} + +bool MessageStorage::RefreshResolvedSnapshot(const AuthorizationSnapshot& snapshot) { + std::lock_guard lock(dbMutex_); + try { + Statement q(db_, "INSERT INTO card_report_state(uid,snapshot,revision,protected) " + "SELECT ?,?,1,1 WHERE NOT EXISTS(SELECT 1 FROM backlog WHERE uid=?) " + "ON CONFLICT(uid) DO UPDATE SET snapshot=excluded.snapshot,protected=1,revision=card_report_state.revision+1 " + "WHERE NOT EXISTS(SELECT 1 FROM backlog WHERE uid=excluded.uid)"); + q.text(1, snapshot.user.uid); q.text(2, snapshotJson(snapshot).dump()); q.text(3, snapshot.user.uid); + return sqlite3_step(q.value) == SQLITE_DONE; + } catch (...) { return false; } +} + +std::vector> MessageStorage::ResolvedSnapshotVersions() const { + std::lock_guard lock(dbMutex_); + std::vector> result; + try { + Statement q(db_, "SELECT uid,revision FROM card_report_state c WHERE protected=1 AND NOT EXISTS(SELECT 1 FROM backlog b WHERE b.uid=c.uid)"); + while (sqlite3_step(q.value) == SQLITE_ROW) + result.emplace_back(columnText(q.value, 0), sqlite3_column_int64(q.value, 1)); + } catch (...) { result.clear(); } + return result; +} + +void MessageStorage::ReleaseResolvedSnapshots(const std::vector>& versions) { + std::lock_guard lock(dbMutex_); + for (const auto& v : versions) { + try { + Statement q(db_, "UPDATE card_report_state SET protected=0,revision=revision+1 WHERE uid=? AND protected=1 AND revision=? AND NOT EXISTS(SELECT 1 FROM backlog WHERE uid=?)"); + q.text(1, v.first); q.number(2, v.second); q.text(3, v.first); + sqlite3_step(q.value); + } catch (...) { return; } + } +} + +std::optional MessageStorage::ClaimNextBacklog() { + std::lock_guard lock(dbMutex_); + try { + Transaction tx(db_); + Statement q(db_, "SELECT id,uid,method,data,attempt,canonical FROM backlog b WHERE in_flight=0 AND retry_after<=CAST(strftime('%s','now') AS INTEGER) AND NOT EXISTS(SELECT 1 FROM backlog earlier WHERE earlier.uid=b.uid AND earlier.id lock(dbMutex_); + try { + Transaction tx(db_); + Statement check(db_, "SELECT uid FROM backlog WHERE id=? AND attempt=? AND in_flight=1"); + check.number(1, m.id); check.number(2, m.attempt); + if (sqlite3_step(check.value) != SQLITE_ROW) return false; + if (result == DeliveryResult::Rejected) { + Statement dead(db_, "INSERT INTO dead_messages(uid,method,data) SELECT uid,method,data FROM backlog WHERE id=?"); + dead.number(1, m.id); + if (sqlite3_step(dead.value) != SQLITE_DONE) return false; + } + Statement change(db_, result == DeliveryResult::Retry + ? "UPDATE backlog SET in_flight=0,retry_after=CAST(strftime('%s','now') AS INTEGER)+? WHERE id=?" + : "DELETE FROM backlog WHERE id=?"); + if (result == DeliveryResult::Retry) { change.number(1, retrySeconds); change.number(2, m.id); } + else change.number(1, m.id); + if (sqlite3_step(change.value) != SQLITE_DONE) return false; + Statement revision(db_, "UPDATE card_report_state SET revision=revision+1 WHERE uid=?"); + revision.text(1, m.uid); + if (sqlite3_step(revision.value) != SQLITE_DONE) return false; + return tx.commit(); + } catch (...) { return false; } +} + +bool MessageStorage::RecoverInFlight() { + return Execute("UPDATE backlog SET in_flight=0,retry_after=0 WHERE in_flight=1"); +} + int MessageStorage::BacklogCount() const { std::lock_guard lock(dbMutex_); if (!db_) { diff --git a/src/state_machine.cpp b/src/state_machine.cpp index 666b913..a1ea05e 100644 --- a/src/state_machine.cpp +++ b/src/state_machine.cpp @@ -73,6 +73,10 @@ bool StateMachine::processEvent(Event event) { if (event == Event::CancelNoFuel) { event = Event::CancelPressed; } + if (event == Event::CancelPressed && getCurrentState() == SystemState::Authorization) { + controller_->cancelSlowAuthorization(); + return true; + } // Coalesce consecutive InputUpdated events to avoid redundant processing/display refreshes. if (event == Event::InputUpdated) { @@ -254,6 +258,7 @@ void StateMachine::setupTransitions() { transitions_[{SystemState::PinEntry, Event::ErrorRecovery}] = {SystemState::PinEntry, noOp}; // From Authorization state + transitions_[{SystemState::Authorization, Event::AuthorizationCancelled}] = {SystemState::Waiting, [this]() { onCancelPressed(); }}; transitions_[{SystemState::Authorization, Event::CardPresented}] = {SystemState::Authorization, noOp}; transitions_[{SystemState::Authorization, Event::PinEntered}] = {SystemState::Authorization, noOp}; transitions_[{SystemState::Authorization, Event::InputUpdated}] = {SystemState::Authorization, noOp}; @@ -609,10 +614,10 @@ DisplayMessage StateMachine::getDisplayMessage() const { break; case SystemState::Authorization: - message.line1 = "Проверка..."; + message.line1 = controller_->isAuthorizationSlow() ? "Медленное соединение" : "Проверка..."; message.line2 = ""; - message.line3 = ""; - message.line4 = "Ожидайте"; + message.line3 = controller_->isAuthorizationSlow() ? "Ожидайте или" : ""; + message.line4 = controller_->isAuthorizationSlow() ? std::string(keyboardUi.cancelPrompt) : "Ожидайте"; break; case SystemState::NotAuthorized: @@ -733,10 +738,10 @@ DisplayMessage StateMachine::getDisplayMessage() const { void StateMachine::doAuthorization() { std::string inputCopy = controller_->getCurrentInput(); - controller_->requestAuthorization(inputCopy); + controller_->beginAuthorization(inputCopy); // Clear sensitive input (PIN/card UID) silently controller_->clearInputSilent(); - // requestAuthorization will post AuthorizationSuccess or AuthorizationFailed event + // The controller accepts this attempt's result on its event loop. } void StateMachine::onAuthorizationSuccess() { @@ -764,7 +769,6 @@ void StateMachine::doRefuelingDataTransmission() { // The session is cleaned up on timeout or other user interactions. if (controller_) { controller_->completeRefueling(); - controller_->postEvent(Event::DataTransmissionComplete); } } @@ -814,7 +818,6 @@ void StateMachine::onIntakeVolumeEntered() { // Do not clear session data here - keep the intake values visible. // The session is cleaned up on timeout or other user interactions. controller_->completeIntakeOperation(); - controller_->postEvent(Event::DataTransmissionComplete); } } diff --git a/src/user_cache.cpp b/src/user_cache.cpp index 4291253..6178595 100644 --- a/src/user_cache.cpp +++ b/src/user_cache.cpp @@ -42,7 +42,7 @@ UserCache::UserCache(const std::string& dbPath) Execute("CREATE TABLE IF NOT EXISTS user_cache_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);"); // Initialize metadata if it doesn't exist - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); sqlite3_stmt* stmt = nullptr; const char* sql = "SELECT value FROM user_cache_meta WHERE key = 'active_table';"; if (sqlite3_prepare_v2(db_, sql, -1, &stmt, nullptr) == SQLITE_OK) { @@ -63,7 +63,7 @@ UserCache::UserCache(const std::string& dbPath) } UserCache::~UserCache() { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); if (db_) { sqlite3_close(db_); db_ = nullptr; @@ -71,12 +71,12 @@ UserCache::~UserCache() { } bool UserCache::IsOpen() const { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); return db_ != nullptr; } bool UserCache::Execute(const std::string& sql) const { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); if (!db_) { return false; } @@ -89,6 +89,21 @@ bool UserCache::Execute(const std::string& sql) const { return true; } +std::optional UserCache::GetAuthorizationSnapshot(const std::string& uid) const { + // Both reads use the same active generation, even during a population flip. + std::lock_guard lock(dbMutex_); + const auto user = GetEntry(uid); + if (!user) return std::nullopt; + AuthorizationSnapshot result; + result.user.uid = user->uid; + result.user.role = static_cast(user->roleId); + result.user.allowance = user->allowance; + for (const auto& tank : GetTanks()) + result.tanks.push_back({tank.idTank, tank.visualNumberTank, tank.nameTank, tank.volume}); + if (result.tanks.empty()) return std::nullopt; + return result; +} + std::string UserCache::GetActiveTableName() const { return activeTableIsA_ ? "user_cache_a" : "user_cache_b"; } @@ -98,7 +113,7 @@ std::string UserCache::GetStandbyTableName() const { } std::optional UserCache::GetEntry(const std::string& uid) const { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); if (!db_) { return std::nullopt; } @@ -124,7 +139,7 @@ std::optional UserCache::GetEntry(const std::string& uid) const } bool UserCache::UpdateEntry(const std::string& uid, double allowance, int roleId) { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); if (!db_) { return false; } @@ -164,7 +179,7 @@ bool UserCache::UpdateEntry(const std::string& uid, double allowance, int roleId } bool UserCache::DeductAllowance(const std::string& uid, double amount) { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); if (!db_) { return false; } @@ -245,7 +260,7 @@ bool UserCache::DeductAllowance(const std::string& uid, double amount) { } int UserCache::GetCount() const { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); if (!db_) { return 0; } @@ -265,7 +280,7 @@ int UserCache::GetCount() const { } std::vector UserCache::GetTanks() const { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); std::vector result; if (!db_) { return result; @@ -293,7 +308,7 @@ std::vector UserCache::GetTanks() const { } int UserCache::GetTankCount() const { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); if (!db_) { return 0; } @@ -314,7 +329,7 @@ int UserCache::GetTankCount() const { } bool UserCache::BeginPopulation() { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); if (!db_ || populationInProgress_) { return false; } @@ -341,7 +356,7 @@ bool UserCache::BeginPopulation() { } bool UserCache::AddPopulationEntry(const std::string& uid, double allowance, int roleId) { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); if (!db_ || !populationInProgress_) { return false; } @@ -362,7 +377,7 @@ bool UserCache::AddPopulationEntry(const std::string& uid, double allowance, int } bool UserCache::AddPopulationTank(int idTank, int visualNumberTank, const std::string& nameTank, double volume) { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); if (!db_ || !populationInProgress_) { return false; } @@ -386,7 +401,7 @@ bool UserCache::AddPopulationTank(int idTank, int visualNumberTank, const std::s } bool UserCache::CommitPopulation() { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); if (!db_ || !populationInProgress_) { return false; } @@ -419,7 +434,7 @@ bool UserCache::CommitPopulation() { } void UserCache::AbortPopulation() { - std::lock_guard lock(dbMutex_); + std::lock_guard lock(dbMutex_); populationInProgress_ = false; } diff --git a/tests/backend_base_test.cpp b/tests/backend_base_test.cpp index 4759991..0d81610 100644 --- a/tests/backend_base_test.cpp +++ b/tests/backend_base_test.cpp @@ -56,6 +56,31 @@ class TestBackendBase : public BackendBase { } // namespace +TEST(BackendBaseReportTest, CanonicalReportsEnforceMethodRoleWithoutRemappingRetries) { + for (int role : {0, 1, 2, 3}) { + TestBackendBase backend("controller"); + int sent = 0; + backend.boolTokenHandler = [&](const std::string& endpoint, const std::string&, + const nlohmann::json& body, bool) -> nlohmann::json { + if (endpoint == "/api/pump/authorize") + return {{"Token", "test"}, {"RoleId", role}, {"Allowance", 1}, + {"fuelTanks", nlohmann::json::array({{{"idTank", 99}, {"visualNumberTank", 7}}})}}; + ++sent; + EXPECT_EQ(body.at("TankNumber"), 42); + EXPECT_EQ(body.at("TimeAt"), 123); + return nullptr; + }; + ASSERT_TRUE(backend.Authorize("card")); + EXPECT_EQ(backend.SendReportPayload(R"({"TankNumber":42,"FuelVolume":10,"TimeAt":123})", false, true), role == 1); + EXPECT_EQ(backend.SendReportPayload(R"({"TankNumber":42,"IntakeVolume":10,"TimeAt":123,"Direction":1})", true, true), role == 2); + EXPECT_EQ(sent, role == 1 || role == 2 ? 1 : 0); + EXPECT_FALSE(backend.SendReportPayload(R"({"TankNumber":42,"FuelVolume":-1,"TimeAt":123})", false, true)); + EXPECT_FALSE(backend.SendReportPayload(R"({"TankNumber":42,"IntakeVolume":10,"TimeAt":123,"Direction":9})", true, true)); + EXPECT_EQ(sent, role == 1 || role == 2 ? 1 : 0); + EXPECT_FALSE(backend.IsNetworkError()); + } +} + TEST(BackendBaseFetchUserCardsTest, SendsExpectedRequestAndParsesValidCards) { TestBackendBase backend("controller-uid-42"); diff --git a/tests/backend_test.cpp b/tests/backend_test.cpp index c2af2e1..6df20c7 100644 --- a/tests/backend_test.cpp +++ b/tests/backend_test.cpp @@ -6,6 +6,7 @@ #include #include "backend.h" #include +#include using namespace fuelflux; using ::testing::_; @@ -464,6 +465,28 @@ TEST_F(BackendTest, ConnectionErrorHandling) { EXPECT_FALSE(backend.GetLastError().empty()); } +TEST_F(BackendTest, CancellationAbortsAnAlreadySentAuthorization) { + std::atomic received{false}, release{false}; + mockServer->handleAuthorize = [&](const httplib::Request&, httplib::Response& response) { + received = true; + while (!release) std::this_thread::sleep_for(std::chrono::milliseconds(5)); + response.set_content(R"({"Token":"late","RoleId":1,"Allowance":100})", "application/json"); + }; + auto backend = std::make_shared(baseAPI, controllerUid); + auto result = std::async(std::launch::async, [backend] { return backend->Authorize("card"); }); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (!received && std::chrono::steady_clock::now() < deadline) + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + EXPECT_TRUE(received); + backend->CancelPendingRequests(); + const auto status = result.wait_for(std::chrono::seconds(2)); + release = true; + EXPECT_EQ(status, std::future_status::ready); + EXPECT_FALSE(result.get()); + EXPECT_FALSE(backend->IsAuthorized()); + EXPECT_TRUE(backend->IsNetworkError()); +} + // Test timeout error handling TEST_F(BackendTest, TimeoutErrorHandling) { // Setup a server that doesn't respond diff --git a/tests/backlog_worker_test.cpp b/tests/backlog_worker_test.cpp index 1c5ab9d..8f1dbd5 100644 --- a/tests/backlog_worker_test.cpp +++ b/tests/backlog_worker_test.cpp @@ -7,6 +7,9 @@ #include "backlog_worker.h" #include "message_storage.h" +#include +#include +#include using namespace fuelflux; using ::testing::Return; @@ -14,6 +17,8 @@ using ::testing::StrictMock; class MockBackendForBacklog : public IBackend { public: + MOCK_METHOD(void, CancelPendingRequests, (), (override)); + MOCK_METHOD(bool, SendReportPayload, (const std::string& payload, bool intake, bool canonicalTankId), (override)); MOCK_METHOD(bool, Authorize, (const std::string& uid), (override)); MOCK_METHOD(bool, Deauthorize, (), (override)); MOCK_METHOD(bool, Refuel, (TankNumber tankNumber, Volume volume), (override)); @@ -38,8 +43,10 @@ TEST(BacklogWorkerTest, ProcessesBacklogSuccessfully) { ASSERT_TRUE(storage->AddBacklog("uid-1", MessageMethod::Refuel, "{\"TankNumber\":1}")); auto backend = std::make_shared>(); + EXPECT_CALL(*backend, CancelPendingRequests()).Times(::testing::AnyNumber()); + EXPECT_CALL(*backend, IsAuthorized()).WillRepeatedly(Return(false)); EXPECT_CALL(*backend, Authorize("uid-1")).WillOnce(Return(true)); - EXPECT_CALL(*backend, RefuelPayload("{\"TankNumber\":1}")).WillOnce(Return(true)); + EXPECT_CALL(*backend, SendReportPayload("{\"TankNumber\":1}", false, false)).WillOnce(Return(true)); EXPECT_CALL(*backend, Deauthorize()).WillOnce(Return(true)); BacklogWorker worker(storage, backend, std::chrono::milliseconds(1)); @@ -53,6 +60,8 @@ TEST(BacklogWorkerTest, KeepsBacklogOnNetworkError) { ASSERT_TRUE(storage->AddBacklog("uid-2", MessageMethod::Refuel, "{\"TankNumber\":2}")); auto backend = std::make_shared>(); + EXPECT_CALL(*backend, CancelPendingRequests()).Times(::testing::AnyNumber()); + EXPECT_CALL(*backend, IsAuthorized()).WillRepeatedly(Return(false)); EXPECT_CALL(*backend, Authorize("uid-2")).WillOnce(Return(false)); EXPECT_CALL(*backend, IsNetworkError()).WillOnce(Return(true)); @@ -67,8 +76,10 @@ TEST(BacklogWorkerTest, MovesToDeadOnNonNetworkError) { ASSERT_TRUE(storage->AddBacklog("uid-3", MessageMethod::Refuel, "{\"TankNumber\":3}")); auto backend = std::make_shared>(); + EXPECT_CALL(*backend, CancelPendingRequests()).Times(::testing::AnyNumber()); + EXPECT_CALL(*backend, IsAuthorized()).WillRepeatedly(Return(false)); EXPECT_CALL(*backend, Authorize("uid-3")).WillOnce(Return(true)); - EXPECT_CALL(*backend, RefuelPayload("{\"TankNumber\":3}")).WillOnce(Return(false)); + EXPECT_CALL(*backend, SendReportPayload("{\"TankNumber\":3}", false, false)).WillOnce(Return(false)); EXPECT_CALL(*backend, Deauthorize()).WillOnce(Return(true)); EXPECT_CALL(*backend, IsNetworkError()).WillOnce(Return(false)); @@ -77,3 +88,46 @@ TEST(BacklogWorkerTest, MovesToDeadOnNonNetworkError) { EXPECT_EQ(storage->BacklogCount(), 0); EXPECT_EQ(storage->DeadMessageCount(), 1); } + +TEST(BacklogWorkerTest, RetriesStartupRecoveryBeforeSendingAnyReport) { + const auto path = std::filesystem::temp_directory_path() / + ("fuelflux-recovery-" + std::to_string(std::random_device{}()) + ".db"); + struct RemoveDatabase { + std::filesystem::path path; + ~RemoveDatabase() { std::error_code error; std::filesystem::remove(path, error); } + } removeDatabase{path}; + auto storage = std::make_shared(path.string()); + ASSERT_TRUE(storage->AddBacklog("first", MessageMethod::Refuel, "first-payload")); + ASSERT_TRUE(storage->ClaimNextBacklog()); + ASSERT_TRUE(storage->AddBacklog("second", MessageMethod::Refuel, "second-payload")); + sqlite3* database = nullptr; + ASSERT_EQ(sqlite3_open(path.string().c_str(), &database), SQLITE_OK); + struct CloseDatabase { sqlite3* db; ~CloseDatabase() { sqlite3_close(db); } } closeDatabase{database}; + ASSERT_EQ(sqlite3_exec(database, + "CREATE TRIGGER block_recovery BEFORE UPDATE ON backlog WHEN OLD.in_flight=1 AND NEW.in_flight=0 BEGIN SELECT RAISE(ABORT,'temporarily unavailable'); END", + nullptr, nullptr, nullptr), SQLITE_OK); + auto backend = std::make_shared>(); + std::atomic sent{0}; + EXPECT_CALL(*backend, CancelPendingRequests()).Times(::testing::AnyNumber()); + { + ::testing::InSequence order; + EXPECT_CALL(*backend, Authorize("first")).WillOnce(Return(true)); + EXPECT_CALL(*backend, SendReportPayload("first-payload", false, false)).WillOnce([&] { ++sent; return true; }); + EXPECT_CALL(*backend, Deauthorize()).WillOnce(Return(true)); + EXPECT_CALL(*backend, Authorize("second")).WillOnce(Return(true)); + EXPECT_CALL(*backend, SendReportPayload("second-payload", false, false)).WillOnce([&] { ++sent; return true; }); + EXPECT_CALL(*backend, Deauthorize()).WillOnce(Return(true)); + } + BacklogWorker worker(storage, backend, std::chrono::milliseconds(10)); + worker.Start(); + std::this_thread::sleep_for(std::chrono::milliseconds(80)); + EXPECT_EQ(sent.load(), 0); + EXPECT_EQ(sqlite3_exec(database, "DROP TRIGGER block_recovery", nullptr, nullptr, nullptr), SQLITE_OK); + worker.Wake(); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (storage->BacklogCount() != 0 && std::chrono::steady_clock::now() < deadline) + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + worker.Stop(); + EXPECT_EQ(sent.load(), 2); + EXPECT_EQ(storage->BacklogCount(), 0); +} diff --git a/tests/cache_manager_test.cpp b/tests/cache_manager_test.cpp index 6c3f19f..a48c607 100644 --- a/tests/cache_manager_test.cpp +++ b/tests/cache_manager_test.cpp @@ -23,6 +23,8 @@ namespace { class MockBackend : public IBackend { public: + MOCK_METHOD(void, CancelPendingRequests, (), (override)); + MOCK_METHOD(bool, SendReportPayload, (const std::string&, bool, bool), (override)); MOCK_METHOD(bool, Authorize, (const std::string& uid), (override)); MOCK_METHOD(bool, Deauthorize, (), (override)); MOCK_METHOD(bool, Refuel, (TankNumber tankNumber, Volume volume), (override)); diff --git a/tests/cares_resolver_test.cpp b/tests/cares_resolver_test.cpp index 9f9feb0..43c2e32 100644 --- a/tests/cares_resolver_test.cpp +++ b/tests/cares_resolver_test.cpp @@ -4,10 +4,18 @@ #include #include "cares_resolver.h" +#include "backend.h" #include +#include +#include #ifdef USE_CARES +#include +#include +#include +#include + namespace fuelflux { namespace { @@ -21,6 +29,7 @@ class CaresEnvironment : public ::testing::Environment { } void TearDown() override { + BackendBase::ShutdownAsyncRequests(); CleanupCaresLibrary(); } }; @@ -37,6 +46,65 @@ class CaresResolverTest : public ::testing::Test { CaresResolver resolver; }; +TEST_F(CaresResolverTest, PreCancelledResolutionDoesNotStart) { + std::atomic cancelled{true}; + const auto start = std::chrono::steady_clock::now(); + EXPECT_TRUE(resolver.Resolve("example.invalid", "", &cancelled).empty()); + EXPECT_LT(std::chrono::steady_clock::now() - start, std::chrono::milliseconds(250)); +} + +TEST_F(CaresResolverTest, CancellationInterruptsResolverLockWait) { + std::atomic hold{false}, entered{false}, release{false}, cancelled{false}; + CaresResolver local("localhost", [&] { + if (hold) { + entered = true; + while (!release) std::this_thread::sleep_for(std::chrono::milliseconds(2)); + } + return CaresResolver::Clock::now(); + }); + ASSERT_FALSE(local.Resolve("localhost").empty()); + hold = true; + auto owner = std::async(std::launch::async, [&] { return local.Resolve("localhost"); }); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (!entered && std::chrono::steady_clock::now() < deadline) + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + auto waiter = std::async(std::launch::async, [&] { return local.Resolve("localhost", "", &cancelled); }); + cancelled = true; + const auto status = waiter.wait_for(std::chrono::milliseconds(500)); + release = true; + EXPECT_TRUE(entered); + EXPECT_EQ(status, std::future_status::ready); + EXPECT_TRUE(waiter.get().empty()); + owner.get(); +} + +TEST_F(CaresResolverTest, CancellationInterruptsPendingDnsResponse) { + // Receive a real DNS query locally, then deliberately withhold the reply. + const int socketFd = socket(AF_INET, SOCK_DGRAM, 0); + ASSERT_GE(socketFd, 0); + struct SocketOwner { int fd; ~SocketOwner() { close(fd); } } owner{socketFd}; + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + ASSERT_EQ(bind(socketFd, reinterpret_cast(&address), sizeof(address)), 0); + socklen_t length = sizeof(address); + ASSERT_EQ(getsockname(socketFd, reinterpret_cast(&address), &length), 0); + CaresResolver local("", CaresResolver::Clock::now, + "127.0.0.1:" + std::to_string(ntohs(address.sin_port))); + std::atomic cancelled{false}; + auto request = std::async(std::launch::async, [&] { + return local.Resolve("foreground-cancellation.invalid", "", &cancelled); + }); + pollfd descriptor{socketFd, POLLIN, 0}; + const int received = poll(&descriptor, 1, 1000); + const auto beforeCancel = request.wait_for(std::chrono::milliseconds(0)); + cancelled = true; + EXPECT_EQ(received, 1); + EXPECT_EQ(beforeCancel, std::future_status::timeout); + EXPECT_EQ(request.wait_for(std::chrono::milliseconds(500)), std::future_status::ready); + EXPECT_TRUE(request.get().empty()); +} + // Test resolving localhost TEST_F(CaresResolverTest, ResolvesLocalhost) { std::string ip = resolver.Resolve("localhost"); diff --git a/tests/controller_test.cpp b/tests/controller_test.cpp index ec184d4..0acdc6b 100644 --- a/tests/controller_test.cpp +++ b/tests/controller_test.cpp @@ -33,6 +33,14 @@ using ::testing::ReturnPointee; using ::testing::ReturnRef; namespace { +::testing::Matcher ReportPayload(int tank, double volume, bool intake) { + return ::testing::Truly([=](const std::string& text) { + const auto payload = nlohmann::json::parse(text, nullptr, false); + return payload.is_object() && payload.value("TankNumber", -1) == tank && + payload.value(intake ? "IntakeVolume" : "FuelVolume", -1.0) == volume && + (!intake || payload.value("Direction", 0) == 1) && payload.contains("TimeAt"); + }); +} std::size_t Utf8CodePointCount(const std::string& text) { std::size_t count = 0; for (unsigned char byte : text) { @@ -83,6 +91,9 @@ using ::testing::NiceMock; // Mock Backend class MockBackend : public IBackend { public: + MOCK_METHOD(void, CancelPendingRequests, (), (override)); + MOCK_METHOD(std::shared_ptr, CreateIndependentSession, (), (const, override)); + MOCK_METHOD(bool, SendReportPayload, (const std::string&, bool, bool), (override)); MOCK_METHOD(bool, Authorize, (const std::string& uid), (override)); MOCK_METHOD(bool, Deauthorize, (), (override)); MOCK_METHOD(bool, Refuel, (TankNumber tankNumber, Volume volume), (override)); @@ -104,6 +115,7 @@ class MockBackend : public IBackend { std::string tokenStorage_; std::vector tanksStorage_; std::string lastErrorStorage_; + std::string controllerUidStorage_; int roleId_ = static_cast(UserRole::Unknown); double allowance_ = 0.0; double price_ = 0.0; @@ -272,7 +284,8 @@ class ControllerTest : public ::testing::Test { void createController(std::chrono::seconds noFlowCancelTimeout = std::chrono::seconds(30)) { auto backend = std::make_shared>(); mockBackend = backend.get(); - ON_CALL(*mockBackend, GetControllerUid()).WillByDefault(ReturnRef(CONTROLLER_UID)); + ON_CALL(*mockBackend, CreateIndependentSession()).WillByDefault([weak = std::weak_ptr(backend)] { return weak.lock(); }); + ON_CALL(*mockBackend, GetControllerUid()).WillByDefault(ReturnRef(mockBackend->controllerUidStorage_)); controller = std::make_unique( CONTROLLER_UID, backend, @@ -299,13 +312,22 @@ class ControllerTest : public ::testing::Test { ON_CALL(*mockCardReader, initialize()).WillByDefault(Return(true)); ON_CALL(*mockPump, initialize()).WillByDefault(Return(true)); ON_CALL(*mockFlowMeter, initialize()).WillByDefault(Return(true)); + EXPECT_CALL(*mockBackend, Authorize(_)).Times(::testing::AnyNumber()).WillRepeatedly([&](const std::string&) { + mockBackend->authorized_ = true; + return true; + }); ON_CALL(*mockBackend, Authorize(_)).WillByDefault([&]() { mockBackend->authorized_ = true; return true; }); ON_CALL(*mockBackend, Refuel(_, _)).WillByDefault(Return(true)); + ON_CALL(*mockBackend, SendReportPayload(_, _, _)).WillByDefault(Return(true)); ON_CALL(*mockBackend, Intake(_, _, _)).WillByDefault(Return(true)); ON_CALL(*mockBackend, IsAuthorized()).WillByDefault(ReturnPointee(&mockBackend->authorized_)); + EXPECT_CALL(*mockBackend, Deauthorize()).Times(::testing::AnyNumber()).WillRepeatedly([&]() { + mockBackend->authorized_ = false; + return true; + }); ON_CALL(*mockBackend, Deauthorize()).WillByDefault([&]() { mockBackend->authorized_ = false; return true; @@ -316,6 +338,12 @@ class ControllerTest : public ::testing::Test { ON_CALL(*mockBackend, GetPrice()).WillByDefault(ReturnPointee(&mockBackend->price_)); ON_CALL(*mockBackend, GetFuelTanks()).WillByDefault(ReturnRef(mockBackend->tanksStorage_)); ON_CALL(*mockBackend, GetLastError()).WillByDefault(ReturnRef(mockBackend->lastErrorStorage_)); + EXPECT_CALL(*mockBackend, FetchUserCards(_, _)) + .Times(::testing::AnyNumber()) + .WillRepeatedly(Return(std::vector{})); + EXPECT_CALL(*mockBackend, FetchFuelTanks(_, _)) + .Times(::testing::AnyNumber()) + .WillRepeatedly(Return(std::vector{})); ON_CALL(*mockDisplay, isConnected()).WillByDefault(Return(true)); ON_CALL(*mockKeyboard, isConnected()).WillByDefault(Return(true)); @@ -364,7 +392,7 @@ class ControllerTest : public ::testing::Test { } bool waitForState(SystemState expected, - std::chrono::milliseconds timeout = std::chrono::milliseconds(500)) { + std::chrono::milliseconds timeout = std::chrono::milliseconds(2000)) { const auto deadline = std::chrono::steady_clock::now() + timeout; while (std::chrono::steady_clock::now() < deadline) { if (controller->getStateMachine().getCurrentState() == expected) { @@ -440,6 +468,8 @@ TEST_F(ControllerTest, PersistencePathsIsolateControllerState) { auto secondBackend = std::make_shared>(); ON_CALL(*secondBackend, GetControllerUid()).WillByDefault(ReturnRef(CONTROLLER_UID)); + ON_CALL(*secondBackend, CreateIndependentSession()) + .WillByDefault([weak = std::weak_ptr(secondBackend)] { return weak.lock(); }); auto secondController = std::make_unique( CONTROLLER_UID, secondBackend, @@ -473,7 +503,7 @@ TEST_F(ControllerTest, AuthorizationFallsBackToCacheOnNetworkError) { ASSERT_TRUE(controller->getUserCache()->AddPopulationTank(10, 7, "Tank-7", 700.0)); ASSERT_TRUE(controller->getUserCache()->CommitPopulation()); - EXPECT_CALL(*mockBackend, Authorize("offline-user")).WillOnce(Return(false)); + EXPECT_CALL(*mockBackend, Authorize("offline-user")).WillRepeatedly(Return(false)); ON_CALL(*mockBackend, IsNetworkError()).WillByDefault(Return(true)); ON_CALL(*mockBackend, FetchUserCards(_, _)).WillByDefault(Return(std::vector{{"offline-user", static_cast(UserRole::Customer), 123.0}})); ON_CALL(*mockBackend, FetchFuelTanks(_, _)).WillByDefault(Return(std::vector{{10, 7, "Tank-7", 700.0}})); @@ -534,7 +564,7 @@ TEST_F(ControllerTest, CachedAuthorizationRefuelGoesToBacklogAndSkipsDeauthorize ASSERT_TRUE(controller->getUserCache()->AddPopulationTank(10, 7, "Tank-7", 700.0)); ASSERT_TRUE(controller->getUserCache()->CommitPopulation()); - EXPECT_CALL(*mockBackend, Authorize("offline-user")).WillOnce(Return(false)); + EXPECT_CALL(*mockBackend, Authorize("offline-user")).WillRepeatedly(Return(false)); ON_CALL(*mockBackend, IsNetworkError()).WillByDefault(Return(true)); ON_CALL(*mockBackend, FetchUserCards(_, _)).WillByDefault(Return(std::vector{{"offline-user", static_cast(UserRole::Customer), 123.0}})); ON_CALL(*mockBackend, FetchFuelTanks(_, _)).WillByDefault(Return(std::vector{{10, 7, "Tank-7", 700.0}})); @@ -572,7 +602,7 @@ TEST_F(ControllerTest, CachedAuthorizationIntakeGoesToBacklog) { ASSERT_TRUE(controller->getUserCache()->AddPopulationTank(20, 11, "Tank-11", 1100.0)); ASSERT_TRUE(controller->getUserCache()->CommitPopulation()); - EXPECT_CALL(*mockBackend, Authorize("offline-operator")).WillOnce(Return(false)); + EXPECT_CALL(*mockBackend, Authorize("offline-operator")).WillRepeatedly(Return(false)); ON_CALL(*mockBackend, IsNetworkError()).WillByDefault(Return(true)); ON_CALL(*mockBackend, FetchUserCards(_, _)).WillByDefault(Return(std::vector{{"offline-operator", static_cast(UserRole::Operator), 0.0}})); ON_CALL(*mockBackend, FetchFuelTanks(_, _)).WillByDefault(Return(std::vector{{20, 11, "Tank-11", 1100.0}})); @@ -594,6 +624,22 @@ TEST_F(ControllerTest, CachedAuthorizationIntakeGoesToBacklog) { EXPECT_EQ(message->method, MessageMethod::Intake); } +TEST_F(ControllerTest, RequestAuthorizationPrefersProtectedSnapshotWhileReportPending) { + AuthorizationSnapshot saved{{"card", UserRole::Customer, 100.0, 0.0}, {{42, 7, "Tank-7", 700.0}}}; + MessageStorage storage(messageStorageDbPath.string()); + ASSERT_TRUE(storage.EnqueueReport(MessageMethod::Refuel, "{}", saved, 10.0)); + + EXPECT_CALL(*mockBackend, Authorize("card")).Times(0); + + controller->requestAuthorization("card"); + + EXPECT_TRUE(controller->isSessionAuthorizedFromCache()); + EXPECT_EQ(controller->getCurrentUser().uid, "card"); + EXPECT_DOUBLE_EQ(controller->getCurrentUser().allowance, 90.0); + ASSERT_EQ(controller->getAvailableTanks().size(), 1u); + EXPECT_EQ(controller->getAvailableTanks()[0].number, 7); +} + // Test Controller initialization TEST_F(ControllerTest, Initialization) { EXPECT_CALL(*mockDisplay, initialize()).Times(1); @@ -943,6 +989,11 @@ TEST_F(ControllerTest, InitializationFailureForcesErrorState) { EXPECT_CALL(*mockCardReader, initialize()).Times(1); EXPECT_CALL(*mockPump, initialize()).Times(1); EXPECT_CALL(*mockFlowMeter, initialize()).Times(1); + EXPECT_CALL(*mockDisplay, shutdown()).Times(1); + EXPECT_CALL(*mockKeyboard, shutdown()).Times(1); + EXPECT_CALL(*mockCardReader, shutdown()).Times(1); + EXPECT_CALL(*mockPump, shutdown()).Times(1); + EXPECT_CALL(*mockFlowMeter, shutdown()).Times(1); bool ok = controller->initialize(); EXPECT_FALSE(ok); @@ -1432,7 +1483,7 @@ TEST_F(ControllerTest, RefuelingCompletionDisplaysFinalVolume) { mockBackend->authorized_ = true; return true; }); - EXPECT_CALL(*mockBackend, Refuel(1, 10.75)).WillOnce(Return(true)); + EXPECT_CALL(*mockBackend, SendReportPayload(ReportPayload(1, 10.75, false), false, true)).WillOnce(Return(true)); EXPECT_CALL(*mockBackend, Deauthorize()).WillOnce([this]() { mockBackend->authorized_ = false; return true; @@ -1598,7 +1649,7 @@ TEST_F(ControllerTest, HandleFlowUpdate) { // This test just verifies the method doesn't crash } -TEST_F(ControllerTest, CalibrationScalesLiveVolumeCutoffAndBackendReportOnce) { +TEST_F(ControllerTest, CalibrationScalesLiveVolumeCutoffAndPersistedReportOnce) { recreateControllerWithCalibration(0.5); mockBackend->roleId_ = static_cast(UserRole::Customer); mockBackend->allowance_ = 100.0; @@ -1616,8 +1667,13 @@ TEST_F(ControllerTest, CalibrationScalesLiveVolumeCutoffAndBackendReportOnce) { EXPECT_DOUBLE_EQ(controller->getCurrentRefuelVolume(), 10.0); EXPECT_FALSE(mockPump->running_); - EXPECT_CALL(*mockBackend, Refuel(1, 10.0)).WillOnce(Return(true)); controller->completeRefueling(); + MessageStorage storage(messageStorageDbPath.string()); + ASSERT_EQ(storage.BacklogCount(), 1); + const auto report = storage.GetNextBacklog(); + ASSERT_TRUE(report); + EXPECT_EQ(nlohmann::json::parse(report->data).at("FuelVolume"), 10.0); + EXPECT_TRUE(report->canonicalTankId); } // Test that rapid handleFlowUpdate calls post InputUpdated at most once per @@ -1935,7 +1991,7 @@ TEST_F(ControllerTest, OperatorIntakeWorkflow) { EXPECT_CALL(*mockBackend, Authorize("operator-card")) .WillOnce(Return(true)); - EXPECT_CALL(*mockBackend, Intake(1, 100.0, IntakeDirection::In)) + EXPECT_CALL(*mockBackend, SendReportPayload(ReportPayload(1, 100.0, true), true, true)) .WillOnce(Return(true)); controller->initialize(); @@ -1973,7 +2029,7 @@ TEST_F(ControllerTest, OperatorIntakeWorkflowSingleTankSkipsSelection) { EXPECT_CALL(*mockBackend, Authorize("operator-card")) .WillOnce(Return(true)); - EXPECT_CALL(*mockBackend, Intake(1, 100.0, IntakeDirection::In)) + EXPECT_CALL(*mockBackend, SendReportPayload(ReportPayload(1, 100.0, true), true, true)) .WillOnce(Return(true)); controller->initialize(); @@ -2008,7 +2064,7 @@ TEST_F(ControllerTest, CustomerRefuelWorkflow) { EXPECT_CALL(*mockBackend, Authorize("customer-card")) .WillOnce(Return(true)); - EXPECT_CALL(*mockBackend, Refuel(1, 50.0)) + EXPECT_CALL(*mockBackend, SendReportPayload(ReportPayload(1, 50.0, false), false, true)) .WillOnce(Return(true)); EXPECT_CALL(*mockBackend, IsAuthorized()) .WillRepeatedly(Return(true)); @@ -2047,7 +2103,7 @@ TEST_F(ControllerTest, CustomerRefuelWorkflowSingleTankSkipsSelection) { EXPECT_CALL(*mockBackend, Authorize("customer-card")) .WillOnce(Return(true)); - EXPECT_CALL(*mockBackend, Refuel(1, 50.0)) + EXPECT_CALL(*mockBackend, SendReportPayload(ReportPayload(1, 50.0, false), false, true)) .WillOnce(Return(true)); EXPECT_CALL(*mockBackend, IsAuthorized()) .WillRepeatedly(Return(true)); @@ -2291,7 +2347,7 @@ TEST_F(ControllerTest, CardReadingDisabledDuringRefueling) { mockBackend->tanksStorage_ = {BackendTankInfo{1, 1, "Tank A"}, BackendTankInfo{2, 2, "Tank B"}}; EXPECT_CALL(*mockBackend, Authorize("test-card")).WillOnce(Return(true)); - EXPECT_CALL(*mockBackend, Refuel(1, 10.0)).WillOnce(Return(true)); + EXPECT_CALL(*mockBackend, SendReportPayload(ReportPayload(1, 10.0, false), false, true)).WillOnce(Return(true)); controller->initialize(); @@ -2351,7 +2407,7 @@ TEST_F(ControllerTest, DataTransmissionStateShownDuringRefuel) { mockBackend->authorized_ = true; return true; }); - EXPECT_CALL(*mockBackend, Refuel(1, 10.0)).WillOnce(Return(true)); + EXPECT_CALL(*mockBackend, SendReportPayload(ReportPayload(1, 10.0, false), false, true)).WillOnce(Return(true)); controller->initialize(); @@ -2416,7 +2472,7 @@ TEST_F(ControllerTest, DataTransmissionStateShownDuringIntake) { mockBackend->authorized_ = true; return true; }); - EXPECT_CALL(*mockBackend, Intake(1, 50.75, IntakeDirection::In)).WillOnce(Return(true)); + EXPECT_CALL(*mockBackend, SendReportPayload(ReportPayload(1, 50.75, true), true, true)).WillOnce(Return(true)); controller->initialize(); diff --git a/tests/foreground_backend_test.cpp b/tests/foreground_backend_test.cpp new file mode 100644 index 0000000..6f4fdd8 --- /dev/null +++ b/tests/foreground_backend_test.cpp @@ -0,0 +1,454 @@ +#include +#include "controller.h" +#include "cache_manager.h" +#include "backend_utils.h" +#include +#include +#include +#include +#include +#include +#include "display/st_bitmap_text.h" +#include "peripherals/keyboard_utils.h" + +using namespace fuelflux; +using namespace std::chrono_literals; +namespace { +struct Network { + std::atomic holdAuth{false}, holdReports{false}, ignoreCancellation{false}; + std::atomic rejectAuth{false}, rejectReports{false}, failReports{false}; + std::atomic tankId{42}; + std::atomic reportCalls{0}, closedSessions{0}; + std::mutex mutex; + std::map authorizations; + std::vector reports; + int AuthCount(const std::string& uid) { + std::lock_guard lock(mutex); return authorizations[uid]; + } +}; + +class TestBackend : public BackendBase { +public: + explicit TestBackend(std::shared_ptr network) + : BackendBase("test-controller", nullptr), network_(std::move(network)) {} + std::shared_ptr CreateIndependentSession() const override { + return std::make_shared(network_); + } +private: + nlohmann::json HttpRequestWrapper(const std::string& endpoint, const std::string&, + const nlohmann::json& body, bool) override { + networkError_ = false; + if (endpoint == "/api/pump/authorize") { + const auto uid = body.at("CardUid").get(); + { std::lock_guard lock(network_->mutex); ++network_->authorizations[uid]; } + while (network_->holdAuth && (network_->ignoreCancellation || !cancelled_)) + std::this_thread::sleep_for(2ms); + if (cancelled_ && !network_->ignoreCancellation) { networkError_ = true; return BuildWrapperErrorResponse(); } + if (network_->rejectAuth) return {{"CodeError", 1}, {"TextError", "Denied"}}; + return {{"Token", "token-" + uid}, {"RoleId", uid == "operator" ? 2 : 1}, {"Allowance", 100.0}, + {"fuelTanks", nlohmann::json::array({{{"idTank", network_->tankId.load()}, {"visualNumberTank", 7}, {"volume", 500.0}}})}}; + } + ++network_->reportCalls; + { std::lock_guard lock(network_->mutex); network_->reports.push_back(body); } + while (network_->holdReports && !cancelled_) std::this_thread::sleep_for(2ms); + if (cancelled_ || network_->failReports) { networkError_ = true; return BuildWrapperErrorResponse(); } + if (network_->rejectReports) return {{"CodeError", 1}, {"TextError", "Rejected"}}; + return nullptr; + } + nlohmann::json HttpRequestWrapper(const std::string& e, const std::string& m, + const nlohmann::json& b, const std::string&) override { + return HttpRequestWrapper(e, m, b, true); + } + void CancelPendingRequests() override { BackendBase::CancelPendingRequests(); } + void SendAsyncDeauthorizeRequest(const std::string&) override { ++network_->closedSessions; } + std::shared_ptr network_; +}; + +class TestPump : public peripherals::IPump { +public: + bool initialize() override { return true; } + void shutdown() override {} + bool isConnected() const override { return true; } + void start() override { running = true; ++starts; } + void stop() override { running = false; } + bool isRunning() const override { return running.load(); } + void setPumpStateCallback(PumpStateCallback) override {} + std::atomic running{false}; + std::atomic starts{0}; +}; + +class TestDisplay : public peripherals::IDisplay { +public: + bool initialize() override { return true; } + void shutdown() override { ++shutdownCalls; } + bool isConnected() const override { return true; } + void showMessage(const DisplayMessage&) override { if (onShow) onShow(); } + void clear() override {} + void setBacklight(bool) override {} + std::function onShow; + std::atomic shutdownCalls{0}; +}; + +class ForegroundBackendTest : public ::testing::Test { +protected: + std::shared_ptr network = std::make_shared(); + std::filesystem::path directory; + std::unique_ptr controller; + std::unique_ptr storage; + TestPump* pump = nullptr; + std::thread thread; + void SetUp() override { + directory = std::filesystem::temp_directory_path() / + ("fuelflux-foreground-" + std::to_string(std::random_device{}())); + std::filesystem::create_directory(directory); + } + void Start(std::chrono::milliseconds threshold = 50ms, bool saved = false, bool startLoop = true) { + // A deliberately unavailable general cache keeps these tests entirely + // local. Protected snapshots exercise the same production fallback path. + std::ofstream(directory / "no-cache") << "not a directory"; + controller = std::make_unique("test-controller", std::make_shared(network), + 30s, ControllerPersistencePaths{(directory / "no-cache" / "cache.db").string(), + (directory / "reports.db").string()}, threshold); + storage = std::make_unique((directory / "reports.db").string()); + auto testPump = std::make_unique(); + pump = testPump.get(); + controller->setPump(std::move(testPump)); + if (saved) Save(); + ASSERT_TRUE(controller->initialize()); + if (startLoop) thread = std::thread([this] { controller->run(); }); + } + void TearDown() override { + network->holdAuth = false; + network->holdReports = false; + if (controller) controller->shutdown(); + if (thread.joinable()) thread.join(); + controller.reset(); storage.reset(); + std::error_code ec; std::filesystem::remove_all(directory, ec); + } + bool Wait(std::function condition, std::chrono::milliseconds timeout = 2000ms) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + if (condition()) return true; + std::this_thread::sleep_for(5ms); + } + return condition(); + } + bool State(SystemState state) { return Wait([&] { return controller->getStateMachine().isInState(state); }); } + void Save(std::string uid = "card") { + AuthorizationSnapshot saved{{uid, UserRole::Customer, 80, 0}, {{42, 7, "Saved tank", 500}}}; + ASSERT_TRUE(storage->EnqueueReport(MessageMethod::Refuel, "{}", saved, 0)); + auto report = storage->ClaimNextBacklog(); + ASSERT_TRUE(report); + ASSERT_TRUE(storage->CompleteDelivery(*report, MessageStorage::DeliveryResult::Accepted)); + } + void Scan(const std::string& uid) { controller->handleCardPresented(uid); } + void Refuel(double volume = 10) { + const auto starts = pump->starts.load(); + controller->enterVolume(volume); + ASSERT_TRUE(State(SystemState::Refueling)); + ASSERT_TRUE(Wait([&] { return pump->starts.load() > starts; })); + controller->handleFlowUpdate(volume); + controller->postEvent(Event::RefuelingStopped); + } +}; +} + +TEST_F(ForegroundBackendTest, AutomaticFallbackDiscardsLateOnlineSuccess) { + Start(350ms, true); network->holdAuth = true; network->ignoreCancellation = true; + const auto started = std::chrono::steady_clock::now(); + Scan("card"); + ASSERT_TRUE(Wait([&] { return network->AuthCount("card") == 1; })); + std::this_thread::sleep_for(100ms); + EXPECT_EQ(controller->getStateMachine().getCurrentState(), SystemState::Authorization); + ASSERT_TRUE(State(SystemState::VolumeEntry)); + EXPECT_GE(std::chrono::steady_clock::now() - started, 350ms); + EXPECT_TRUE(controller->isSessionAuthorizedFromCache()); + EXPECT_DOUBLE_EQ(controller->getCurrentUser().allowance, 80); + network->holdAuth = false; + ASSERT_TRUE(Wait([&] { return network->closedSessions.load() == 1; })); + EXPECT_EQ(controller->getStateMachine().getCurrentState(), SystemState::VolumeEntry); + EXPECT_DOUBLE_EQ(controller->getCurrentUser().allowance, 80); + EXPECT_DOUBLE_EQ(storage->GetProtectedSnapshot("card")->authorization.user.allowance, 80); +} + +TEST_F(ForegroundBackendTest, MissingCacheShowsWarningAndCancelAllowsDifferentCard) { + Start(350ms); network->holdAuth = true; network->ignoreCancellation = true; + const auto started = std::chrono::steady_clock::now(); + Scan("unknown"); + ASSERT_TRUE(Wait([&] { return network->AuthCount("unknown") == 1; })); + std::this_thread::sleep_for(100ms); + EXPECT_FALSE(controller->isAuthorizationSlow()); + ASSERT_TRUE(Wait([&] { return controller->isAuthorizationSlow(); })); + EXPECT_GE(std::chrono::steady_clock::now() - started, 350ms); + const auto display = controller->getStateMachine().getDisplayMessage(); + EXPECT_EQ(display.line1, "Медленное соединение"); + EXPECT_EQ(display.line3, "Ожидайте или"); + EXPECT_NE(display.line4.find("ОТМЕНА"), std::string::npos); + controller->postEvent(Event::CancelPressed); + ASSERT_TRUE(State(SystemState::Waiting)); + network->holdAuth = false; + Scan("next"); + ASSERT_TRUE(State(SystemState::VolumeEntry)); + EXPECT_EQ(controller->getCurrentUser().uid, "next"); + EXPECT_FALSE(controller->isSessionAuthorizedFromCache()); + ASSERT_TRUE(Wait([&] { return network->closedSessions.load() >= 1; })); + EXPECT_EQ(controller->getCurrentUser().uid, "next"); +} + +TEST_F(ForegroundBackendTest, ConfiguredThresholdAlsoControlsReportReleaseAndSameCardUsesCache) { + Start(350ms); + Scan("card"); ASSERT_TRUE(State(SystemState::VolumeEntry)); + EXPECT_FALSE(controller->isSessionAuthorizedFromCache()); + network->holdReports = true; + const auto started = std::chrono::steady_clock::now(); + Refuel(); + ASSERT_TRUE(State(SystemState::RefuelingComplete)); + EXPECT_GE(std::chrono::steady_clock::now() - started, 350ms); + ASSERT_EQ(network->reportCalls.load(), 1); + EXPECT_TRUE(storage->HasPendingReports("card")); + Scan("card"); ASSERT_TRUE(State(SystemState::VolumeEntry)); + EXPECT_TRUE(controller->isSessionAuthorizedFromCache()); + EXPECT_EQ(network->AuthCount("card"), 1); + EXPECT_DOUBLE_EQ(controller->getCurrentUser().allowance, 90); + network->holdReports = false; + ASSERT_TRUE(Wait([&] { return !storage->HasPendingReports("card"); })); + EXPECT_EQ(controller->getStateMachine().getCurrentState(), SystemState::VolumeEntry); + EXPECT_DOUBLE_EQ(controller->getCurrentUser().allowance, 90); + EXPECT_EQ(network->reportCalls.load(), 1); + { std::lock_guard lock(network->mutex); EXPECT_EQ(network->reports[0].at("TankNumber"), 42); } + controller->postEvent(Event::CancelPressed); ASSERT_TRUE(State(SystemState::Waiting)); + Scan("card"); ASSERT_TRUE(State(SystemState::VolumeEntry)); + EXPECT_FALSE(controller->isSessionAuthorizedFromCache()); + EXPECT_EQ(network->AuthCount("card"), 2); +} + +TEST_F(ForegroundBackendTest, LateReportRejectionDoesNotAffectDifferentCard) { + Start(); Scan("card"); ASSERT_TRUE(State(SystemState::VolumeEntry)); + network->holdReports = true; network->rejectReports = true; + Refuel(); ASSERT_TRUE(State(SystemState::RefuelingComplete)); + Scan("other"); ASSERT_TRUE(State(SystemState::VolumeEntry)); + EXPECT_FALSE(controller->isSessionAuthorizedFromCache()); + network->holdReports = false; + ASSERT_TRUE(Wait([&] { return storage->DeadMessageCount() == 1; })); + EXPECT_EQ(controller->getCurrentUser().uid, "other"); + EXPECT_FALSE(storage->HasPendingReports("card")); + controller->postEvent(Event::CancelPressed); ASSERT_TRUE(State(SystemState::Waiting)); + Scan("card"); ASSERT_TRUE(State(SystemState::VolumeEntry)); + EXPECT_FALSE(controller->isSessionAuthorizedFromCache()); +} + +TEST_F(ForegroundBackendTest, ShutdownCancelsBlockedAuthorization) { + Start(); network->holdAuth = true; Scan("card"); + ASSERT_TRUE(Wait([&] { return network->AuthCount("card") == 1; })); + const auto started = std::chrono::steady_clock::now(); + EXPECT_TRUE(controller->shutdown()); + thread.join(); + EXPECT_LT(std::chrono::steady_clock::now() - started, 1500ms); + EXPECT_FALSE(controller->isSessionAuthorizedFromCache()); +} + +TEST_F(ForegroundBackendTest, ShutdownDeauthorizesActiveOnlineSession) { + Start(); + Scan("card"); + ASSERT_TRUE(State(SystemState::VolumeEntry)); + EXPECT_EQ(network->AuthCount("card"), 1); + + EXPECT_TRUE(controller->shutdown()); + thread.join(); + + ASSERT_TRUE(Wait([&] { return network->closedSessions.load() == 1; })); +} + +TEST_F(ForegroundBackendTest, ShutdownDuringReportingRestoresPendingCardAllowance) { + Start(); Scan("card"); ASSERT_TRUE(State(SystemState::VolumeEntry)); + network->holdReports = true; + Refuel(); ASSERT_TRUE(State(SystemState::RefuelingComplete)); + ASSERT_EQ(network->reportCalls.load(), 1); + EXPECT_TRUE(controller->shutdown()); + thread.join(); + controller.reset(); storage.reset(); + Start(); Scan("card"); ASSERT_TRUE(State(SystemState::VolumeEntry)); + EXPECT_TRUE(controller->isSessionAuthorizedFromCache()); + EXPECT_DOUBLE_EQ(controller->getCurrentUser().allowance, 90); + EXPECT_TRUE(storage->HasPendingReports("card")); +} + +TEST_F(ForegroundBackendTest, ShutdownBeforeRunPreventsQueuedPeripheralActions) { + Start(50ms, false, false); + auto display = std::make_unique(); + std::atomic calls{0}; + display->onShow = [&] { ++calls; }; + controller->setDisplay(std::move(display)); + controller->postEvent(Event::FlowDisplayRefresh); + std::promise launch; + auto ready = launch.get_future(); + thread = std::thread([&, ready = std::move(ready)]() mutable { ready.wait(); controller->run(); }); + EXPECT_TRUE(controller->shutdown()); + launch.set_value(); + thread.join(); + EXPECT_EQ(calls.load(), 0); +} + +TEST_F(ForegroundBackendTest, UnexpectedLoopExitStillAllowsShutdown) { + Start(50ms, false, false); + auto display = std::make_unique(); + auto* displayPtr = display.get(); + display->onShow = [] { throw std::runtime_error("display failure"); }; + controller->setDisplay(std::move(display)); + std::atomic exited{false}; + thread = std::thread([&] { + try { controller->run(); } catch (const std::runtime_error&) { exited = true; } + }); + controller->postEvent(Event::FlowDisplayRefresh); + EXPECT_TRUE(Wait([&] { return exited.load(); })); + EXPECT_TRUE(controller->shutdown()); + thread.join(); + EXPECT_EQ(displayPtr->shutdownCalls.load(), 1); +} + +TEST_F(ForegroundBackendTest, ShutdownDeadlinePreservesLiveStateAndAllowsRetry) { + Start(50ms, false, false); + auto display = std::make_unique(); + auto* displayPtr = display.get(); + std::promise release; + auto released = release.get_future().share(); + std::atomic entered{false}; + display->onShow = [&, released] { entered = true; released.wait(); }; + controller->setDisplay(std::move(display)); + thread = std::thread([this] { controller->run(); }); + controller->postEvent(Event::FlowDisplayRefresh); + EXPECT_TRUE(Wait([&] { return entered.load(); })); + const auto started = std::chrono::steady_clock::now(); + EXPECT_FALSE(controller->shutdown()); + EXPECT_LT(std::chrono::steady_clock::now() - started, timing::kShutdownDeadline + 1s); + EXPECT_EQ(displayPtr->shutdownCalls.load(), 0); + release.set_value(); + thread.join(); + EXPECT_TRUE(controller->shutdown()); + EXPECT_EQ(displayPtr->shutdownCalls.load(), 1); +} + +TEST_F(ForegroundBackendTest, SynchronousOnlyBackendIsRejectedBeforeControllerUse) { + class SynchronousBackend : public TestBackend { + public: + using TestBackend::TestBackend; + std::shared_ptr CreateIndependentSession() const override { return {}; } + }; + EXPECT_THROW(Controller("controller", std::make_shared(network), 30s, + ControllerPersistencePaths{(directory / "cache.db").string(), (directory / "reports.db").string()}), + std::invalid_argument); +} + +TEST_F(ForegroundBackendTest, PendingReportsAccumulateDeductionsAndLateRepliesDoNotRestoreThem) { + Start(); Scan("card"); ASSERT_TRUE(State(SystemState::VolumeEntry)); + network->holdReports = true; + Refuel(10); ASSERT_TRUE(State(SystemState::RefuelingComplete)); + Scan("card"); ASSERT_TRUE(State(SystemState::VolumeEntry)); + Refuel(5); ASSERT_TRUE(State(SystemState::RefuelingComplete)); + EXPECT_EQ(network->reportCalls.load(), 1); + Scan("card"); ASSERT_TRUE(State(SystemState::VolumeEntry)); + EXPECT_DOUBLE_EQ(controller->getCurrentUser().allowance, 85); + EXPECT_EQ(network->AuthCount("card"), 1); + network->holdReports = false; + ASSERT_TRUE(Wait([&] { return !storage->HasPendingReports("card"); })); + EXPECT_EQ(network->reportCalls.load(), 2); + EXPECT_DOUBLE_EQ(controller->getCurrentUser().allowance, 85); + EXPECT_DOUBLE_EQ(storage->GetProtectedSnapshot("card")->authorization.user.allowance, 85); +} + +TEST_F(ForegroundBackendTest, EarlyDenialDoesNotUseSavedData) { + Start(350ms, true); network->rejectAuth = true; + Scan("card"); ASSERT_TRUE(State(SystemState::NotAuthorized)); + EXPECT_FALSE(controller->isSessionAuthorizedFromCache()); + EXPECT_FALSE(controller->isAuthorizationSlow()); +} + +TEST_F(ForegroundBackendTest, CancelBeforeThresholdCannotCancelLaterSuccess) { + Start(350ms); network->holdAuth = true; + Scan("card"); ASSERT_TRUE(Wait([&] { return network->AuthCount("card") == 1; })); + controller->postEvent(Event::CancelPressed); + network->holdAuth = false; + ASSERT_TRUE(State(SystemState::VolumeEntry)); + EXPECT_FALSE(controller->isSessionAuthorizedFromCache()); +} + +TEST_F(ForegroundBackendTest, SlowIntakeReleasesScreenAndRetainsCanonicalPayload) { + Start(); Scan("operator"); ASSERT_TRUE(State(SystemState::IntakeDirectionSelection)); + controller->handleKeyPress(KeyCode::Key1); + controller->handleKeyPress(KeyCode::KeyStart); + ASSERT_TRUE(State(SystemState::IntakeVolumeEntry)); + network->holdReports = true; + controller->enterIntakeVolume(12.5); + ASSERT_TRUE(State(SystemState::IntakeComplete)); + Scan("operator"); ASSERT_TRUE(State(SystemState::IntakeDirectionSelection)); + EXPECT_TRUE(controller->isSessionAuthorizedFromCache()); + network->holdReports = false; + ASSERT_TRUE(Wait([&] { return !storage->HasPendingReports("operator"); })); + EXPECT_EQ(controller->getStateMachine().getCurrentState(), SystemState::IntakeDirectionSelection); + std::lock_guard lock(network->mutex); + ASSERT_EQ(network->reports.size(), 1u); + EXPECT_EQ(network->reports[0].at("IntakeVolume"), 12.5); + EXPECT_EQ(network->reports[0].at("TankNumber"), 42); +} + +TEST_F(ForegroundBackendTest, FailedPersistenceShowsStorageErrorWithoutReleasingTransaction) { + Start(); Scan("card"); ASSERT_TRUE(State(SystemState::VolumeEntry)); + sqlite3* db = nullptr; + ASSERT_EQ(sqlite3_open((directory / "reports.db").string().c_str(), &db), SQLITE_OK); + ASSERT_EQ(sqlite3_exec(db, "CREATE TRIGGER fail_report BEFORE INSERT ON backlog BEGIN SELECT RAISE(ABORT,'full'); END", nullptr, nullptr, nullptr), SQLITE_OK); + sqlite3_close(db); + Refuel(); + ASSERT_TRUE(State(SystemState::Error)); + EXPECT_EQ(controller->getLastErrorMessage(), "Ошибка записи"); + EXPECT_EQ(storage->BacklogCount(), 0); + ASSERT_TRUE(storage->GetProtectedSnapshot("card")); + EXPECT_DOUBLE_EQ(storage->GetProtectedSnapshot("card")->authorization.user.allowance, 100.0); + EXPECT_EQ(network->reportCalls.load(), 0); +} + +TEST_F(ForegroundBackendTest, UnwritableStorageRejectsAuthorizationBeforeBackendCall) { + Start(); + sqlite3* db = nullptr; + ASSERT_EQ(sqlite3_open((directory / "reports.db").string().c_str(), &db), SQLITE_OK); + ASSERT_EQ(sqlite3_exec(db, "CREATE TRIGGER fail_write BEFORE UPDATE ON device_settings BEGIN SELECT RAISE(ABORT,'full'); END", nullptr, nullptr, nullptr), SQLITE_OK); + sqlite3_close(db); + + Scan("card"); + + ASSERT_TRUE(State(SystemState::CannotAuthorize)); + EXPECT_EQ(network->AuthCount("card"), 0); + EXPECT_EQ(controller->getLastErrorMessage(), "Ошибка записи"); +} + +TEST(ForegroundDisplayTest, WarningAndBothCancelLabelsFitSmallDisplay) { + fuelflux::display::StBitmapText font(fuelflux::display::StBitmapFontSize::Small6x12); + EXPECT_EQ(font.fittedGlyphCount("Медленное соединение", 124), 20u); + for (const auto type : {fuelflux::peripherals::KeyboardType::Legacy, fuelflux::peripherals::KeyboardType::Vid}) { + const std::string prompt(fuelflux::peripherals::keyboardUiProfile(type).cancelPrompt); + const auto count = font.fittedGlyphCount(prompt, 124); + EXPECT_EQ(count, type == fuelflux::peripherals::KeyboardType::Vid ? 14u : 18u); + } + EXPECT_GT(timing::kForegroundBackendWaitTimeout.count(), 0); +} + +TEST(ReportWorkerTest, NetworkRetryKeepsCanonicalTankTimestampAndSingleDeduction) { + auto network = std::make_shared(); + auto storage = std::make_shared(":memory:"); + auto backend = std::make_shared(network); + BacklogWorker worker(storage, backend, 1ms); + AuthorizationSnapshot snapshot{{"card", UserRole::Customer, 100, 0}, {{42, 7, "Saved tank", 500}}}; + const std::string payload = R"({"TankNumber":42,"FuelVolume":10,"TimeAt":123456})"; + ASSERT_TRUE(worker.Submit(MessageMethod::Refuel, payload, snapshot, 10)); + network->failReports = true; + EXPECT_FALSE(worker.ProcessOnce()); + EXPECT_TRUE(storage->HasPendingReports("card")); + network->failReports = false; + network->tankId = 99; + EXPECT_TRUE(worker.ProcessOnce()); + EXPECT_FALSE(storage->HasPendingReports("card")); + EXPECT_DOUBLE_EQ(storage->GetProtectedSnapshot("card")->authorization.user.allowance, 90); + std::lock_guard lock(network->mutex); + ASSERT_EQ(network->reports.size(), 2u); + EXPECT_EQ(network->reports[0], nlohmann::json::parse(payload)); + EXPECT_EQ(network->reports[1], network->reports[0]); +} diff --git a/tests/report_delivery_test.cpp b/tests/report_delivery_test.cpp new file mode 100644 index 0000000..1687ebe --- /dev/null +++ b/tests/report_delivery_test.cpp @@ -0,0 +1,189 @@ +#include +#include "message_storage.h" +#include +#include +#include + +using namespace fuelflux; +namespace { +AuthorizationSnapshot SavedCard(std::string uid = "card", double allowance = 100.0) { + return {{std::move(uid), UserRole::Customer, allowance, 0.0}, {{42, 7, "Tank", 500.0}}}; +} +struct TempDatabase { + std::filesystem::path path = std::filesystem::temp_directory_path() / + ("fuelflux-delivery-" + std::to_string(std::random_device{}()) + ".db"); + ~TempDatabase() { std::error_code ec; std::filesystem::remove(path, ec); } +}; +} + +TEST(ReportDeliveryTest, RetryKeepsPayloadAndDeductsOnlyOnce) { + MessageStorage storage(":memory:"); + auto id = storage.EnqueueReport(MessageMethod::Refuel, "{\"TankNumber\":42,\"TimeAt\":123}", SavedCard(), 10); + ASSERT_TRUE(id); + auto card = storage.GetProtectedSnapshot("card"); + ASSERT_TRUE(card); + EXPECT_TRUE(card->pending); + EXPECT_DOUBLE_EQ(card->authorization.user.allowance, 90); + auto first = storage.ClaimNextBacklog(); + ASSERT_TRUE(first); + EXPECT_TRUE(first->canonicalTankId); + EXPECT_FALSE(storage.ClaimNextBacklog()); + ASSERT_TRUE(storage.CompleteDelivery(*first, MessageStorage::DeliveryResult::Retry, 0)); + auto retry = storage.ClaimNextBacklog(); + ASSERT_TRUE(retry); + EXPECT_GT(retry->attempt, first->attempt); + EXPECT_EQ(retry->data, first->data); + EXPECT_FALSE(storage.CompleteDelivery(*first, MessageStorage::DeliveryResult::Accepted)); + EXPECT_TRUE(storage.HasPendingReports("card")); + ASSERT_TRUE(storage.CompleteDelivery(*retry, MessageStorage::DeliveryResult::Accepted)); + EXPECT_FALSE(storage.HasPendingReports("card")); + EXPECT_DOUBLE_EQ(storage.GetProtectedSnapshot("card")->authorization.user.allowance, 90); + auto second = storage.EnqueueReport(MessageMethod::Refuel, "{}", card->authorization, 5); + ASSERT_TRUE(second); + EXPECT_GT(*second, *id); + EXPECT_DOUBLE_EQ(storage.GetProtectedSnapshot("card")->authorization.user.allowance, 85); +} + +TEST(ReportDeliveryTest, AtomicRollbackWhenReportInsertFails) { + TempDatabase file; + MessageStorage storage(file.path.string()); + sqlite3* db = nullptr; + ASSERT_EQ(sqlite3_open(file.path.string().c_str(), &db), SQLITE_OK); + ASSERT_EQ(sqlite3_exec(db, "CREATE TRIGGER fail_insert BEFORE INSERT ON backlog BEGIN SELECT RAISE(ABORT,'full'); END", nullptr, nullptr, nullptr), SQLITE_OK); + EXPECT_FALSE(storage.EnqueueReport(MessageMethod::Refuel, "{}", SavedCard(), 10)); + EXPECT_FALSE(storage.GetProtectedSnapshot("card")); + EXPECT_EQ(storage.BacklogCount(), 0); + sqlite3_close(db); +} + +TEST(ReportDeliveryTest, AuthorizationSnapshotIsImmutableAndPrefersPendingCardData) { + MessageStorage storage(":memory:"); + auto general = SavedCard("card", 500); + ASSERT_TRUE(storage.EnqueueReport(MessageMethod::Refuel, "{}", SavedCard(), 10)); + const auto captured = storage.CaptureAuthorizationState("card", [&] { return general; }); + ASSERT_TRUE(captured.saved); + EXPECT_TRUE(captured.reportStorageAvailable); + EXPECT_TRUE(captured.pendingReports); + EXPECT_DOUBLE_EQ(captured.saved->user.allowance, 90); + storage.EnqueueReport(MessageMethod::Refuel, "{}", *captured.saved, 5); + general.user.allowance = 1000; + EXPECT_DOUBLE_EQ(captured.saved->user.allowance, 90); + EXPECT_DOUBLE_EQ(storage.CaptureAuthorizationState("card", [&] { return general; }).saved->user.allowance, 85); +} + +TEST(ReportDeliveryTest, AuthorizationSnapshotRequiresWritableReportStorage) { + TempDatabase file; + MessageStorage storage(file.path.string()); + sqlite3* db = nullptr; + ASSERT_EQ(sqlite3_open(file.path.string().c_str(), &db), SQLITE_OK); + ASSERT_EQ(sqlite3_exec(db, "CREATE TRIGGER fail_write BEFORE UPDATE ON device_settings BEGIN SELECT RAISE(ABORT,'full'); END", nullptr, nullptr, nullptr), SQLITE_OK); + const auto captured = storage.CaptureAuthorizationState("card", [] { return SavedCard(); }); + EXPECT_FALSE(captured.reportStorageAvailable); + EXPECT_FALSE(captured.saved); + sqlite3_close(db); +} + +TEST(ReportDeliveryTest, PerCardOrderAndOtherCardsCanProgress) { + MessageStorage storage(":memory:"); + auto a = storage.EnqueueReport(MessageMethod::Refuel, "{}", SavedCard(), 10); + storage.EnqueueReport(MessageMethod::Refuel, "{}", SavedCard("card", 90), 10); + auto b = storage.EnqueueReport(MessageMethod::Intake, "{}", SavedCard("other"), 0); + auto first = storage.ClaimNextBacklog(); + ASSERT_TRUE(first); EXPECT_EQ(first->id, a); + auto other = storage.ClaimNextBacklog(); + ASSERT_TRUE(other); EXPECT_EQ(other->id, b); + EXPECT_FALSE(storage.ClaimNextBacklog()); + ASSERT_TRUE(storage.CompleteDelivery(*first, MessageStorage::DeliveryResult::Rejected)); + EXPECT_EQ(storage.DeadMessageCount(), 1); + EXPECT_TRUE(storage.HasPendingReports("card")); + auto next = storage.ClaimNextBacklog(); + ASSERT_TRUE(next); + ASSERT_TRUE(storage.CompleteDelivery(*next, MessageStorage::DeliveryResult::Rejected)); + EXPECT_FALSE(storage.HasPendingReports("card")); +} + +TEST(ReportDeliveryTest, RestartInvalidatesOldAttemptsAndPreservesBalance) { + TempDatabase file; + StoredMessage old; + { + MessageStorage storage(file.path.string()); + ASSERT_TRUE(storage.EnqueueReport(MessageMethod::Refuel, "{}", SavedCard(), 12)); + old = *storage.ClaimNextBacklog(); + } + { + MessageStorage storage(file.path.string()); + EXPECT_FALSE(storage.ClaimNextBacklog()); + ASSERT_TRUE(storage.RecoverInFlight()); + EXPECT_FALSE(storage.CompleteDelivery(old, MessageStorage::DeliveryResult::Accepted)); + ASSERT_TRUE(storage.RecoverInFlight()); + auto resumed = storage.ClaimNextBacklog(); + ASSERT_TRUE(resumed); + EXPECT_EQ(resumed->id, old.id); + EXPECT_EQ(resumed->attempt, old.attempt + 1); + EXPECT_FALSE(storage.CompleteDelivery(old, MessageStorage::DeliveryResult::Accepted)); + EXPECT_DOUBLE_EQ(storage.GetProtectedSnapshot("card")->authorization.user.allowance, 88); + } +} + +TEST(ReportDeliveryTest, StaleSynchronizationCannotReleaseLocalAllowance) { + MessageStorage storage(":memory:"); + storage.EnqueueReport(MessageMethod::Refuel, "{}", SavedCard(), 10); + EXPECT_TRUE(storage.ResolvedSnapshotVersions().empty()); + auto first = storage.ClaimNextBacklog(); + ASSERT_TRUE(first); + storage.CompleteDelivery(*first, MessageStorage::DeliveryResult::Accepted); + const auto versions = storage.ResolvedSnapshotVersions(); + ASSERT_EQ(versions.size(), 1u); + storage.ClearProtectedSnapshot("card"); + storage.EnqueueReport(MessageMethod::Refuel, "{}", SavedCard("card", 90), 10); + auto next = storage.ClaimNextBacklog(); + storage.CompleteDelivery(*next, MessageStorage::DeliveryResult::Accepted); + storage.ReleaseResolvedSnapshots(versions); + ASSERT_TRUE(storage.GetProtectedSnapshot("card")); + EXPECT_DOUBLE_EQ(storage.GetProtectedSnapshot("card")->authorization.user.allowance, 80); + storage.ReleaseResolvedSnapshots(storage.ResolvedSnapshotVersions()); + EXPECT_FALSE(storage.GetProtectedSnapshot("card")); +} + +TEST(ReportDeliveryTest, MigratesLegacyPayloadsAndOrder) { + TempDatabase file; + sqlite3* db = nullptr; + ASSERT_EQ(sqlite3_open(file.path.string().c_str(), &db), SQLITE_OK); + ASSERT_EQ(sqlite3_exec(db, "CREATE TABLE backlog(uid TEXT,method TEXT,data TEXT); INSERT INTO backlog(rowid,uid,method,data) VALUES(4,'a','Refuel','legacy-a'),(9,'b','Intake','legacy-b')", nullptr, nullptr, nullptr), SQLITE_OK); + sqlite3_close(db); + MessageStorage storage(file.path.string()); + auto a = storage.ClaimNextBacklog(); + ASSERT_TRUE(a); EXPECT_EQ(a->id, 4); EXPECT_EQ(a->data, "legacy-a"); EXPECT_FALSE(a->canonicalTankId); + auto b = storage.ClaimNextBacklog(); + ASSERT_TRUE(b); EXPECT_EQ(b->id, 9); EXPECT_EQ(b->data, "legacy-b"); + storage.CompleteDelivery(*a, MessageStorage::DeliveryResult::Accepted); + storage.CompleteDelivery(*b, MessageStorage::DeliveryResult::Accepted); + ASSERT_TRUE(storage.AddBacklog("c", MessageMethod::Refuel, "new")); + EXPECT_GT(storage.GetNextBacklog()->id, 9); +} + +TEST(ReportDeliveryTest, FreshAuthorizationCannotExposeOlderSynchronizationData) { + MessageStorage storage(":memory:"); + storage.EnqueueReport(MessageMethod::Refuel, "{}", SavedCard(), 10); + const auto report = storage.ClaimNextBacklog(); + ASSERT_TRUE(report); + storage.CompleteDelivery(*report, MessageStorage::DeliveryResult::Accepted); + const auto oldSync = storage.ResolvedSnapshotVersions(); + ASSERT_TRUE(storage.RefreshResolvedSnapshot(SavedCard("card", 70))); + storage.ReleaseResolvedSnapshots(oldSync); + ASSERT_TRUE(storage.GetProtectedSnapshot("card")); + EXPECT_DOUBLE_EQ(storage.GetProtectedSnapshot("card")->authorization.user.allowance, 70); + EXPECT_FALSE(storage.HasPendingReports("card")); + storage.ReleaseResolvedSnapshots(storage.ResolvedSnapshotVersions()); + EXPECT_FALSE(storage.GetProtectedSnapshot("card")); +} + +TEST(ReportDeliveryTest, RefreshResolvedSnapshotCreatesProtectionWithoutPriorPendingState) { + MessageStorage storage(":memory:"); + EXPECT_FALSE(storage.GetProtectedSnapshot("card")); + ASSERT_TRUE(storage.RefreshResolvedSnapshot(SavedCard("card", 70))); + ASSERT_TRUE(storage.GetProtectedSnapshot("card")); + EXPECT_DOUBLE_EQ(storage.GetProtectedSnapshot("card")->authorization.user.allowance, 70); + storage.ReleaseResolvedSnapshots(storage.ResolvedSnapshotVersions()); + EXPECT_FALSE(storage.GetProtectedSnapshot("card")); +}