Refactoring to support dynamic loading of backtesting sweeps

CMakeLists.txt

@@ -93,6 +93,20 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON)
 # module flags for loadCommand.cppm — are always present for the editor.
 set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
 
+# Clang 22 made "reduced BMI" the default for C++20 modules, and reduced BMIs
+# drop the predeclared global operator new/delete from the global module
+# (llvm/llvm-project#166068). A TU that imports std alongside enough of this
+# project's modules then segfaults clang's codegen (EmitBuiltinNewDeleteCall,
+# release build's llvm_unreachable) while emitting __builtin_operator_new from
+# libc++'s __libcpp_allocate — main.cpp was the victim. Force full BMIs (the
+# pre-22 behaviour) until the upstream fix reaches a release. Global
+# CMAKE_CXX_FLAGS on purpose: it must also reach CMake's internal `import std`
+# target (see the OpenMP note above the flags append near the bottom).
+if(CMAKE_CXX_COMPILER_ID STREQUAL "Clang"
+   AND CMAKE_CXX_COMPILER_VERSION VERSION_GREATER_EQUAL 22)
+    string(APPEND CMAKE_CXX_FLAGS " -fno-modules-reduced-bmi")
+endif()
+
 # Configure libpqxx build
 set(PQXX_LIBRARIES_INSTALL ON)
 set(SKIP_BUILD_TEST ON)
@@ -198,7 +212,9 @@ string(APPEND CMAKE_CXX_FLAGS " ${OpenMP_CXX_FLAGS}")
 # compiled into one dedicated TU via <boost/redis/src.hpp>
 # (source/shared/redis/connection/boostRedisImpl.cpp), which isolates its
 # macro/SSL pollution from the rest of the build. It pulls in Boost.Asio's SSL
-# layer, which needs OpenSSL.
+# layer, which needs OpenSSL. Boost.Interprocess (the shared-memory control
+# channel, source/shared/ipc/engineControl.cpp) is also header-only; on Linux it
+# needs librt for shm_open (glibc < 2.34), which is a no-op elsewhere.
 find_package(Boost REQUIRED CONFIG)
 find_package(OpenSSL REQUIRED)
 find_package(Threads REQUIRED)
@@ -208,7 +224,8 @@ target_link_libraries(BacktestingEngineLib PUBLIC
     OpenSSL::SSL
     OpenSSL::Crypto
     Threads::Threads
-    CURL::libcurl)
+    CURL::libcurl
+    $<$<PLATFORM_ID:Linux>:rt>)
 
 add_subdirectory(external/boost-decimal)
 target_link_libraries(BacktestingEngineLib PUBLIC Boost::decimal)

LICENSE.MD

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2024 Ryan McCaffery
+Copyright (c) 2026 Ryan McCaffery https://mccaffers.com
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

scripts/ingest.sh

@@ -1,11 +1,11 @@
 #!/bin/bash
 # Builds the engine and runs the `ingest` subcommand: it binds a UDP socket and
-# streams decoded ticks (tickPacket) into the converted QuestDB via ILP-over-HTTP
+# streams decoded ticks (tickPacket) into QuestDB via ILP-over-HTTP
 # (source/ingest/ingestCommand.cppm). Blocks until SIGINT/SIGTERM.
 #
 # Config is read from the environment by IngestCommand (defaults in parens):
 #   QUESTDB_HOST      QuestDB host                (127.0.0.1)
-#   QUESTDB_ILP_PORT  converted-DB ILP/HTTP port  (9001)
+#   QUESTDB_ILP_PORT  QuestDB ILP/HTTP port       (9000)
 #   INGEST_BIND_ADDR  UDP bind address            (127.0.0.1)
 #   INGEST_UDP_PORT   UDP bind port               (11111, == UDPPorts.PortSave)
 #
@@ -30,7 +30,7 @@ fi
 # QuestDB is down (the writer sheds once its buffer fills), so warn rather than
 # abort.
 quest_host="${QUESTDB_HOST:-127.0.0.1}"
-quest_port="${QUESTDB_ILP_PORT:-9001}"
+quest_port="${QUESTDB_ILP_PORT:-9000}"
 # Probe a real endpoint (/exec), not "/": QuestDB's web console root can stall
 # indefinitely, which made a 1s timeout report a healthy DB as unreachable.
 if ! curl --fail --silent --max-time 2 "http://$quest_host:$quest_port/exec?query=SELECT%201" >/dev/null 2>&1; then

scripts/load.sh

@@ -15,9 +15,9 @@ if [ ! -f "$BUILD_DIR/$EXECUTABLE_NAME" ]; then
     exit 1
 fi
 
-if ! redis-cli -h localhost ping >/dev/null 2>&1; then
-    echo "redis-server not reachable on localhost:6379 — aborting"
-    exit 1
+if ! redis-cli -h "$REDIS_HOST" ping >/dev/null 2>&1; then
+    echo "redis-server not reachable on $REDIS_HOST:6379 — skipping"
+    exit 0
 fi
 
 exec ./"$BUILD_DIR/$EXECUTABLE_NAME" load

scripts/run.sh

@@ -31,8 +31,8 @@ else
     exit 1
 fi
 
-if ! redis-cli -h localhost ping >/dev/null 2>&1; then
-    echo "redis-server not reachable on localhost:6379 — skipping"
+if ! redis-cli -h "$REDIS_HOST" ping >/dev/null 2>&1; then
+    echo "redis-server not reachable on $REDIS_HOST:6379 — skipping"
     exit 0
 fi
 

source/ingest/ingestCommand.cppm

@@ -14,6 +14,8 @@
 
 module;
 
+#include <ctime>  // POSIX gmtime_r (not exported by `import std`)
+
 #include "ingest/questdbIngestClient.hpp"
 #include "ingest/udpPorts.hpp"
 #include "ingest/udpReceiver.hpp"
@@ -43,14 +45,40 @@ std::uint16_t parsePort(std::string_view text, std::uint16_t fallback) {
     return static_cast<std::uint16_t>(value);
 }
 
+// UTC wall-clock prefix, e.g. "[2026-06-28 15:04:20.785]", to millisecond
+// precision. gmtime_r is POSIX (hence the <ctime> include in the global module
+// fragment) — the same approach tradeManager.cppm uses for UTC tick times.
+std::string timestamp() {
+    const auto now = std::chrono::system_clock::now();
+    const std::time_t t = std::chrono::system_clock::to_time_t(now);
+    const auto millis = std::chrono::duration_cast<std::chrono::milliseconds>(
+                            now.time_since_epoch())
+                            .count() %
+                        1000;
+    std::tm utc{};
+    gmtime_r(&t, &utc);
+    return std::format("[{:04}-{:02}-{:02} {:02}:{:02}:{:02}.{:03}]",
+                       utc.tm_year + 1900, utc.tm_mon + 1, utc.tm_mday,
+                       utc.tm_hour, utc.tm_min, utc.tm_sec, millis);
+}
+
+// Like std::println to stdout, but timestamped and flushed immediately. stdout
+// is fully buffered when redirected (pipe/file/service log), so without the
+// flush the once-a-minute reports below would sit in the buffer and only appear
+// in a burst when the process exits.
+template <class... Args>
+void logLine(std::format_string<Args...> fmt, Args&&... args) {
+    std::println("{} {}", timestamp(), std::format(fmt, std::forward<Args>(args)...));
+    std::fflush(stdout);
+}
+
 }  // namespace
 
 int IngestCommand::run(const int argc, const char* argv[]) {
     const std::string questHost = env::getOr("QUESTDB_HOST", "127.0.0.1");
-    // 9001 is the converted (INT32) QuestDB's HTTP/ILP port — the instance the
-    // backtester reads and that convert.py writes to. (9000 is the legacy DOUBLE
-    // source DB.) Override with $QUESTDB_ILP_PORT.
-    const std::uint16_t ilpPort = parsePort(env::getOr("QUESTDB_ILP_PORT", ""), 9001);
+    // QuestDB serves ILP-over-HTTP (POST /write) on its main HTTP port, 9000 by
+    // default. Override with $QUESTDB_ILP_PORT.
+    const std::uint16_t ilpPort = parsePort(env::getOr("QUESTDB_ILP_PORT", ""), 9000);
     const std::string bindAddr = env::getOr("INGEST_BIND_ADDR", "127.0.0.1");
 
     // Bind port: CLI arg (argv[2]) overrides $INGEST_UDP_PORT overrides kSave.
@@ -82,16 +110,52 @@ int IngestCommand::run(const int argc, const char* argv[]) {
             received.fetch_add(1, std::memory_order_relaxed);
         });
 
-    std::println("IngestCommand: starting; binding udp://{}:{} -> QuestDB ILP http://{}:{}/write",
-                 bindAddr, bindPort, questHost, ilpPort);
-
-    if (!receiver.run()) {  // binds the socket, then blocks until SIGINT/SIGTERM
-        return 1;           // bind failed (bad address / port in use) — logged
+    logLine("IngestCommand: starting; binding udp://{}:{} -> QuestDB ILP http://{}:{}/write",
+            bindAddr, bindPort, questHost, ilpPort);
+
+    // receiver.run() blocks the main thread, so a background thread prints
+    // throughput once a minute. It wakes every second to notice shutdown
+    // promptly. Plain std::thread + atomic flag on purpose — std::jthread's
+    // stop_token wait (condition_variable_any) doesn't link under `import std`
+    // here (see module-migration notes).
+    std::atomic<bool> reporting{true};
+    std::thread reporter([&] {
+        using clock = std::chrono::steady_clock;
+        constexpr auto interval = std::chrono::minutes(1);
+        std::uint64_t lastReceived = 0;
+        auto nextReport = clock::now() + interval;
+        while (reporting.load(std::memory_order_relaxed)) {
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+            if (clock::now() < nextReport) {
+                continue;
+            }
+            nextReport += interval;
+            const auto total = received.load(std::memory_order_relaxed);
+            const auto delta = total - lastReceived;
+            lastReceived = total;
+            logLine("IngestCommand: received={} (+{}, ~{}/s); "
+                    "decode-dropped={}, queue-dropped={}, post-failed={}",
+                    total, delta, delta / 60,
+                    dropped.load(std::memory_order_relaxed), writer.droppedLines(),
+                    writer.failedLines());
+        }
+    });
+
+    const bool ok = receiver.run();  // binds the socket, then blocks until SIGINT/SIGTERM
+
+    reporting.store(false, std::memory_order_relaxed);
+    reporter.join();
+
+    if (!ok) {
+        return 1;  // bind failed (bad address / port in use) — logged
     }
 
     // decode-dropped: bad size / unknown symbol. queue-dropped: shed by the
     // writer when QuestDB couldn't keep up and the buffer hit its cap.
-    std::println("IngestCommand: shutting down (received={}, decode-dropped={}, queue-dropped={})",
-                 received.load(), dropped.load(), writer.droppedLines());
+    // post-failed: lines a POST couldn't persist (transport error / non-2xx).
+    logLine("IngestCommand: shutting down (received={}, decode-dropped={}, "
+            "queue-dropped={}, post-failed={})",
+            received.load(), dropped.load(), writer.droppedLines(),
+            writer.failedLines());
     return 0;
 }

source/ingest/questdbIngestClient.cpp

@@ -12,7 +12,9 @@
 #include "ingest/questdbIngestClient.hpp"
 
 #include <algorithm>
+#include <atomic>
 #include <condition_variable>
+#include <csignal>
 #include <deque>
 #include <mutex>
 #include <stop_token>
@@ -33,18 +35,37 @@
 
 void ensureCurlInit() {
     static const struct CurlGlobal {
-        CurlGlobal() { curl_global_init(CURL_GLOBAL_ALL); }
+        CurlGlobal() {
+            // Ignore SIGPIPE process-wide. The worker POSTs with CURLOPT_NOSIGNAL
+            // (so libcurl no longer guards SIGPIPE itself); without this, a write
+            // to a QuestDB connection that was reset mid-transfer would raise
+            // SIGPIPE, whose default action kills the process. Asio's signal_set
+            // only handles SIGINT/SIGTERM, so nothing else catches it.
+            std::signal(SIGPIPE, SIG_IGN);
+            curl_global_init(CURL_GLOBAL_ALL);
+        }
         ~CurlGlobal() { curl_global_cleanup(); }
     } guard;
     (void)guard;
 }
 
-// Swallow the response body so curl does not dump it to stdout (its default when
-// no write callback is set). QuestDB replies 204 on success, 400 + a JSON body
-// describing the offending line on a parse/type error.
-std::size_t discardResponse(char* /*ptr*/, std::size_t size, std::size_t nmemb,
-                            void* /*userdata*/) {
-    return size * nmemb;
+// Capture the response body (bounded) into the std::string at `userdata`, so a
+// non-2xx reply can be logged. QuestDB replies 204 with an empty body on
+// success, or 400 + a JSON body naming the offending line on a parse/type error
+// — that body is the only place the reason appears. We keep at most
+// kMaxResponseLog bytes so a large/hostile reply can't bloat the log line, but
+// still tell curl we consumed everything (without a write callback curl would
+// dump the body to stdout).
+constexpr std::size_t kMaxResponseLog = 1024;
+
+std::size_t captureResponse(char* ptr, std::size_t size, std::size_t nmemb,
+                            void* userdata) {
+    const std::size_t n = size * nmemb;
+    auto* body = static_cast<std::string*>(userdata);
+    if (body->size() < kMaxResponseLog) {
+        body->append(ptr, std::min(n, kMaxResponseLog - body->size()));
+    }
+    return n;
 }
 
 }  // namespace
@@ -63,6 +84,8 @@
                                            // link bug, and this is a classic TU
     std::deque<std::string> queue;
     std::size_t droppedTotal = 0;          // guarded by mtx
+    std::atomic<std::size_t> failedTotal{0};  // POST/HTTP failures; worker writes,
+                                              // reporter reads — atomic, no mtx
 
     CURL* curl = nullptr;                  // owned by the worker thread only
     std::jthread worker;
@@ -109,31 +132,45 @@
     // the queue was empty (nothing sent).
     bool drainAndPost() {
         std::string batch;
+        std::size_t lineCount = 0;
         {
             std::unique_lock lock(mtx);
             const std::size_t n = std::min(queue.size(), batchSize);
             for (std::size_t i = 0; i < n; ++i) {
                 batch += std::move(queue.front());
                 queue.pop_front();
             }
+            lineCount = n;
         }
         if (batch.empty()) {
             return false;
         }
-        post(batch);
+        post(batch, lineCount);
         return true;
     }
 
-    void post(const std::string& body) {
+    // POST one batch body holding `lineCount` lines. On any failure (init,
+    // transport error, or non-2xx) those lines did not persist, so they are
+    // added to failedTotal.
+    void post(const std::string& body, std::size_t lineCount) {
         if (!curl) {
+            failedTotal.fetch_add(lineCount, std::memory_order_relaxed);
             return;  // init failed earlier; error already logged
         }
+        std::string responseBody;
         curl_easy_reset(curl);
         curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
         curl_easy_setopt(curl, CURLOPT_POST, 1L);
         curl_easy_setopt(curl, CURLOPT_POSTFIELDS, body.data());
         curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, static_cast<long>(body.size()));
-        curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, discardResponse);
+        curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, captureResponse);
+        curl_easy_setopt(curl, CURLOPT_WRITEDATA, &responseBody);
+        // Run signal-free: this POST happens on the worker thread, and libcurl's
+        // default signal-based (SIGALRM/siglongjmp) timeout path is not
+        // thread-safe and races the main thread's Asio signal_set. NOSIGNAL also
+        // suppresses libcurl's SIGPIPE handling — we ignore SIGPIPE globally in
+        // ensureCurlInit() to compensate.
+        curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L);
         // Bound blocking time so a dead/slow QuestDB can't wedge the worker.
         curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, kConnectTimeoutSeconds);
         curl_easy_setopt(curl, CURLOPT_TIMEOUT, kTransferTimeoutSeconds);
@@ -142,13 +179,15 @@
         if (rc != CURLE_OK) {
             backtest_log::error(std::string("QuestdbIngestClient: POST failed: ")
                                 + curl_easy_strerror(rc));
+            failedTotal.fetch_add(lineCount, std::memory_order_relaxed);
             return;
         }
         long status = 0;
         curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &status);
         if (status < 200 || status >= 300) {
             backtest_log::error("QuestdbIngestClient: HTTP " + std::to_string(status)
-                                + " from " + url);
+                                + " from " + url + ": " + responseBody);
+            failedTotal.fetch_add(lineCount, std::memory_order_relaxed);
         }
     }
 };
@@ -194,4 +233,8 @@
     return impl_->droppedTotal;
 }
 
+std::size_t QuestdbIngestClient::failedLines() const {
+    return impl_->failedTotal.load(std::memory_order_relaxed);
+}
+
 }  // namespace ingest

source/ingest/questdbIngestClient.hpp

@@ -31,7 +31,7 @@ class QuestdbIngestClient {
     // when QuestDB is slow/down — past it the oldest lines are dropped so memory
     // stays bounded rather than growing without limit.
     explicit QuestdbIngestClient(std::string host,
-                                 std::uint16_t port = 9001,
+                                 std::uint16_t port = 9000,
                                  std::size_t batchSize = 1000,
                                  std::chrono::milliseconds flushInterval =
                                      std::chrono::milliseconds(100),
@@ -50,6 +50,11 @@ class QuestdbIngestClient {
     // keep up). Thread-safe.
     [[nodiscard]] std::size_t droppedLines() const;
 
+    // Total lines that failed to persist because a POST errored (transport
+    // failure / timeout) or QuestDB returned a non-2xx status. Distinct from
+    // droppedLines (those never left the queue). Thread-safe.
+    [[nodiscard]] std::size_t failedLines() const;
+
 private:
     struct Impl;
     std::unique_ptr<Impl> impl_;