feat(dc_bridge): extract the Uploader into its own dc_uploader process

Closes #446

The Uploader used to run on a worker thread inside dc_bridge, configured by
ROS parameters and linking the AWS SDK into the Bridge's own process. It is
now dc_uploader, a standalone executable configured entirely by DC_UPLOADER_*
environment variables, with no ROS dependency. The Bridge keeps only the
Files subscription and the durable intent-queue write side; dc_uploader reads
that queue, uploads, and emits the resulting status Records itself.

IntentQueue gains rescan() so a reader-only instance in a separate process can
discover intents a different process enqueued — the queue's on-disk format
was already crash-atomic and safe for this, but each process's in-memory
scheduling state previously only ever loaded what was on disk at construction
time. dc_bringup.launch.py starts dc_uploader automatically alongside
dc_bridge (same ExecuteProcess pattern as dc_mcap_writer) whenever a
`receives: files` Destination is configured, translating that Destination
plus files.*/uploader.data_dir into dc_uploader's environment — deployments
keep the same params file.

See docs/adr/0014-uploader-runs-as-its-own-process.md for the full decision.
Verified with a full tools/e2e/scripts/build.sh run (colcon build + test):
656 tests, 0 failures, and a live smoke test of dc_uploader against RustFS.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VXhyy2nX5JokxtTnGdwzjy
Signed-off-by: David Bensoussan <d.bensoussan@proton.me>

Author: Minipada

dc_bridge/CMakeLists.txt

@@ -33,8 +33,8 @@ find_package(AWSSDK REQUIRED COMPONENTS s3)
 
 # ROS-independent core: Forwarder/Supervisor/Readiness/TopicConfig/ConfigRenderer/
 # vector_binary plus the Uploader logic (ADR-0005). Deliberately aws-free — the S3
-# ObjectStore implementation lives in the node target below — so the Uploader is
-# unit-tested with an in-memory fake and plain gtest, no cloud dependency.
+# ObjectStore implementation lives in the dc_uploader executable below (#446) — so the
+# Uploader is unit-tested with an in-memory fake and plain gtest, no cloud dependency.
 add_library(dc_bridge_core
   src/forwarder.cpp
   src/readiness.cpp
@@ -52,7 +52,8 @@ add_library(dc_bridge_core
   src/uploader/intent_queue.cpp
   src/uploader/thumbnail.cpp
   src/uploader/uploader.cpp
-  src/uploader/retention.cpp)
+  src/uploader/retention.cpp
+  src/uploader/process_config.cpp)
 target_include_directories(dc_bridge_core PUBLIC
   $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>
   $<INSTALL_INTERFACE:include>)
@@ -79,16 +80,26 @@ target_link_libraries(dc_bridge_ros dc_bridge_core)
 ament_target_dependencies(dc_bridge_ros
   rclcpp rosidl_runtime_cpp rosidl_typesupport_introspection_cpp rcpputils)
 
-# The ROS node, plus the aws-sdk-cpp S3 ObjectStore implementation (the only Uploader
-# piece that links the SDK).
+# The ROS node. No longer links the AWS SDK (#446): uploading moved to dc_uploader below,
+# so the ROS container never needs to hold object-storage credentials (epic #440's user
+# story 5) and the Bridge's own link line stays aws-free.
 add_executable(dc_bridge
   src/main.cpp
-  src/bridge_node.cpp
-  src/uploader/s3_object_store.cpp)
-target_link_libraries(dc_bridge dc_bridge_core dc_bridge_ros ${AWSSDK_LINK_LIBRARIES})
+  src/bridge_node.cpp)
+target_link_libraries(dc_bridge dc_bridge_core dc_bridge_ros)
 ament_target_dependencies(dc_bridge rclcpp dc_interfaces std_msgs std_srvs)
 
-install(TARGETS dc_bridge DESTINATION lib/${PROJECT_NAME})
+# The Uploader's own entrypoint (#446): reads the durable intent queue the Bridge above
+# writes, uploads Files to object storage, and emits each File's metadata Record itself —
+# a standalone process, configured entirely by DC_UPLOADER_* environment variables
+# (dc_bridge/uploader/process_config.hpp), with no ROS dependency. Links the aws-sdk-cpp
+# S3 ObjectStore implementation, the one Uploader piece that needs it.
+add_executable(dc_uploader
+  src/uploader_main.cpp
+  src/uploader/s3_object_store.cpp)
+target_link_libraries(dc_uploader dc_bridge_core ${AWSSDK_LINK_LIBRARIES})
+
+install(TARGETS dc_bridge dc_uploader DESTINATION lib/${PROJECT_NAME})
 
 if(BUILD_TESTING)
   find_package(ament_cmake_gtest REQUIRED)
@@ -97,7 +108,7 @@ if(BUILD_TESTING)
   # dc_bridge itself needs no part of it — the Bridge is deliberately unaware of incident capture.
   find_package(dc_common REQUIRED)
   # All gtests link only the aws-free core (uploader_test uses an in-memory ObjectStore).
-  foreach(t forwarder supervisor render misc uploader intent_queue retention thumbnail raw_config)
+  foreach(t forwarder supervisor render misc uploader intent_queue retention thumbnail raw_config process_config)
     ament_add_gtest(${t}_test test/${t}_test.cpp)
     target_link_libraries(${t}_test dc_bridge_core)
     ament_target_dependencies(${t}_test dc_common)

dc_bridge/README.md

@@ -10,16 +10,24 @@ from an in-memory unacked window, so a Record is only forgotten once the Shipper
 actually durably buffered it — see "Delivery guarantees" below. It renders Vector's
 configuration from ROS parameters for the blessed Destination set (`postgres`, `s3`,
 `file`, `console` — ADR-0003) and passes raw Vector snippets (`custom_config_files`)
-through for everything else. It also hosts the
-**Uploader** (ADR-0005): `receives: files` Destinations are served by the Bridge itself —
-Records referencing Files get their Files uploaded to S3-compatible object storage
-(multipart + resumable), verified, and reported as status/metadata Records under the
-`dc.files` Tag. Pending uploads are durable (#265): each is written to a disk-backed
-intent queue before being handed to the Uploader, so a Bridge restart replays whatever
-never got acked instead of forgetting it. That queue and the Uploader's multipart-resume
-state live under `uploader.data_dir` (#441), a directory separate from the Shipper's own
-`shipper.data_dir` buffer — they default to the same path, so a deployment that only ever
-set `shipper.data_dir` keeps working unchanged, but each can be mounted as its own volume.
+through for everything else.
+
+This package also builds **`dc_uploader`** (ADR-0005, docs/adr/0014-uploader-runs-as-its-
+own-process.md): a separate, ROS-free executable that serves `receives: files`
+Destinations — Records referencing Files get their Files uploaded to S3-compatible object
+storage (multipart + resumable), verified, and reported as status/metadata Records under
+the `dc.files` Tag. The Bridge's own job for a files Destination is only to subscribe and
+durably enqueue (#265): each Record referencing a File is written to a disk-backed intent
+queue before the callback returns, so a Bridge *or* `dc_uploader` restart replays whatever
+never got acked instead of forgetting it. `dc_uploader` is configured entirely by
+`DC_UPLOADER_*` environment variables, not ROS parameters — see `process_config.hpp`. The
+queue and `dc_uploader`'s multipart-resume state live under `uploader.data_dir` (#441), a
+directory separate from the Shipper's own `shipper.data_dir` buffer — they default to the
+same path, so a deployment that only ever set `shipper.data_dir` keeps working unchanged,
+but each can be mounted as its own volume. `dc_bringup.launch.py` starts `dc_uploader`
+alongside `dc_bridge` automatically whenever a `receives: files` Destination is
+configured, translating that Destination plus `files.*`/`uploader.data_dir` into
+`dc_uploader`'s environment — no separate manual step.
 
 `dc_bridge` is an ordinary `ament_cmake` C++ package. It builds with the same `rosdep
 install` + `colcon build` as every other `dc_*` package. See
@@ -46,26 +54,31 @@ install` + `colcon build` as every other `dc_*` package. See
   - `atomic_write` — writes a file crash/partial-write-safe (write to `<path>.tmp`, then
     `rename()`, #444), used for the rendered Shipper config so a reader polling the path
     never observes a partial write.
-  - `uploader/` — the Uploader (ADR-0005): `group` (parses the Files a Record
+  - `uploader/` — the Uploader's logic (ADR-0005): `group` (parses the Files a Record
     references), `content_type` (magic-byte sniffing), `status` (the Humble-compatible
     status-row shapes), `multipart` (resumable multipart with a per-part JSON checkpoint
     sidecar), `intent_queue` (#265 — the disk-backed durable queue of pending uploads:
     crash-atomic tmp+rename enqueue, ack-by-unlink, oldest-first sweep with per-entry
-    exponential backoff, replayed on startup), and `uploader` (the verify-then-delete
-    orchestration). All built on an abstract `ObjectStore` interface, so the logic is
-    tested against an in-memory fake with no cloud dependency.
+    exponential backoff, replayed on startup; `rescan()`, #446, is how a second process's
+    own instance over the same directory learns about an intent it didn't enqueue
+    itself), `uploader` (the verify-then-delete orchestration), and `process_config`
+    (#446 — parses `dc_uploader`'s `DC_UPLOADER_*` environment-variable configuration).
+    All built on an abstract `ObjectStore` interface, so the logic is tested against an
+    in-memory fake with no cloud dependency.
 - **`src/raw_subscriptions.cpp`** (`dc_bridge_ros`) — raw / generic-subscription mode's
   rclcpp half (#227): topic discovery, `create_generic_subscription`, and the runtime
   introspection walk that turns a message of a type the Bridge was never compiled against
   into JSON. Its own library rather than a file in the executable, so the conversion is
   gtested against real message types without the node, Vector or the AWS SDK.
 - **`src/uploader/s3_object_store.cpp`** — the aws-sdk-cpp implementation of
-  `ObjectStore` (the only Uploader piece that links the SDK, via `aws_sdk_vendor`).
-  Verified against RustFS (PutObject + multipart) before adoption.
+  `ObjectStore`, linked only into `dc_uploader` (the only Uploader piece that links the
+  SDK, via `aws_sdk_vendor`) — `dc_bridge` itself no longer does. Verified against RustFS
+  (PutObject + multipart) before adoption.
 - **`src/bridge_node.cpp` / `src/main.cpp`** — the `rclcpp` node: declares the
-  `shipper`/`uploader`/`destinations`/`files` parameters (ADR-0003 config contract + ADR-0005),
-  renders and atomically writes the config (write then rename, #444), subscribes to every
-  Destination's `inputs` topics, forwards Records, runs the Uploader worker thread, and
+  `shipper`/`uploader`/`destinations`/`files.metadata_destination` parameters (ADR-0003
+  config contract + ADR-0005), renders and atomically writes the config (write then
+  rename, #444), subscribes to every Destination's `inputs` topics, forwards Records,
+  writes upload intents to the durable queue for `dc_uploader` to read (#446), and
   exposes a `~/ready` (`std_srvs/Trigger`) service. In the default **managed** mode
   (`shipper.managed: true`) it also locates the vendored Vector binary, `vector
   validate`s the merged config, and spawns/supervises Vector. In **unmanaged** mode
@@ -77,8 +90,14 @@ install` + `colcon build` as every other `dc_*` package. See
   probe against the Shipper's ingest port, independent of who spawned it. `rclcpp` handles
   SIGINT/SIGTERM and `on_shutdown` stops Vector in managed mode (a no-op in unmanaged
   mode, nothing to stop), so it's never orphaned.
+- **`src/uploader_main.cpp`** — `dc_uploader`'s entry point (#446, no `rclcpp`): loads
+  `DC_UPLOADER_*` configuration, builds the S3 `ObjectStore`, and runs the same
+  upload/retention poll loop the Bridge's worker thread used to run, now against its own
+  `IntentQueue` instance (kept in sync with the Bridge's via `rescan()`). SIGINT/SIGTERM
+  set an atomic flag the loop checks every poll.
 - **`test/`** — gtest suites (`forwarder`, `supervisor`, `render`, `misc`, `uploader`,
-  `intent_queue`), all linking only the aws-free core.
+  `intent_queue`, `retention`, `thumbnail`, `raw_config`, `process_config`), all linking
+  only the aws-free core.
 
 ## Config renderer (ADR-0003)
 

dc_bridge/include/dc_bridge/bridge_node.hpp

@@ -19,12 +19,13 @@
 // Records-Destination's `inputs` topics and forwards each Record to Vector over the
 // shipper ingest protocol.
 //
-// `receives: files` Destinations (ADR-0005, the Uploader) are handled in Phase 2.
+// `receives: files` Destinations (ADR-0005): the Bridge subscribes and writes upload
+// intents to the durable queue; a separate dc_uploader process (#446) reads, uploads, and
+// emits the resulting status Records — see docs/adr/0014-uploader-runs-as-its-own-process.md.
 #ifndef DC_BRIDGE__BRIDGE_NODE_HPP_
 #define DC_BRIDGE__BRIDGE_NODE_HPP_
 
 #include <atomic>
-#include <condition_variable>
 #include <memory>
 #include <mutex>
 #include <nlohmann/json.hpp>
@@ -40,10 +41,8 @@
 #include "dc_bridge/readiness.hpp"
 #include "dc_bridge/render.hpp"
 #include "dc_bridge/supervisor.hpp"
+#include "dc_bridge/uploader/file_status_tag.hpp"
 #include "dc_bridge/uploader/intent_queue.hpp"
-#include "dc_bridge/uploader/object_store.hpp"
-#include "dc_bridge/uploader/retention.hpp"
-#include "dc_bridge/uploader/uploader.hpp"
 #include "dc_interfaces/msg/string_stamped.hpp"
 
 namespace dc_bridge
@@ -62,11 +61,6 @@ class BridgeNode : public rclcpp::Node
 
 private:
   void run_prober(std::string forward_host, std::uint16_t forward_port);
-  // The Uploader worker (ADR-0005/#265): sweeps the durable intent queue oldest-first
-  // (per-entry backoff on failure — one permanently-failing intent can't starve the
-  // backlog), processes each Record, acks (unlinks) its intent on success, and emits
-  // status Records through its own Forwarder connection under FILE_STATUS_TAG.
-  void run_uploader_worker(std::string forward_host, std::uint16_t forward_port);
 
   std::shared_ptr<Supervisor> supervisor_;
   std::mutex supervisor_mutex_;
@@ -82,28 +76,12 @@ class BridgeNode : public rclcpp::Node
   std::atomic<bool> prober_stop_{ false };
   std::atomic<bool> stopped_{ false };
 
-  // Uploader (ADR-0005) — created only when a `receives: files` destination is
-  // configured. Records on files-destination topics are enqueued into the durable
-  // intent_queue_ (#265); the worker thread replays/sweeps it, uploads Files, and emits
-  // status Records.
-  std::unique_ptr<uploader::Uploader> uploader_;
+  // The durable upload intent queue (ADR-0005/#265) — created only when a `receives:
+  // files` destination is configured. Records on files-destination topics are enqueued
+  // here so they survive a Bridge crash/restart. The Bridge only writes; a separate
+  // dc_uploader process (#446) owns reading, replaying, uploading, and emitting status
+  // Records — see docs/adr/0014-uploader-runs-as-its-own-process.md.
   std::unique_ptr<uploader::IntentQueue> intent_queue_;
-  std::thread uploader_thread_;
-  std::mutex uploader_wake_mutex_;
-  std::condition_variable uploader_wake_cv_;
-  bool upload_stop_{ false };
-
-  // Files retention (#267) — a copy of the storages the Uploader was built with (kept
-  // alongside it since the Uploader moves its own copy) so the worker thread can build
-  // shed audit rows without needing an Uploader accessor; a no-op sweep when disabled
-  // (the default) either way.
-  uploader::RetentionConfig retention_config_;
-  std::vector<Storage> files_storages_;
-
-  // Thumbnails (#256) — the configured generator binary, kept for the uploader worker's
-  // "couldn't generate any previews" warning, which is the only place a failed preview
-  // is observable (it never fails the upload itself).
-  std::string thumbnail_binary_;
 
   // Raw / generic-subscription mode (#227) — created only when `raw.enabled` is true.
   // Its Records go out through the same Forwarder as the Measurement Records above,

dc_bridge/include/dc_bridge/uploader/file_status_tag.hpp

@@ -0,0 +1,18 @@
+// SPDX-FileCopyrightText: 2022-2026 David Bensoussan
+// SPDX-License-Identifier: MPL-2.0
+
+// Split out of uploader.hpp (#446) so a process that only needs to know the Uploader's
+// Tag — the Bridge, wiring Vector's routing for it — doesn't have to pull in the full
+// Uploader/ObjectStore/Retention class surface just for one constant.
+#ifndef DC_BRIDGE__UPLOADER__FILE_STATUS_TAG_HPP_
+#define DC_BRIDGE__UPLOADER__FILE_STATUS_TAG_HPP_
+
+namespace dc_bridge::uploader
+{
+
+/// The Tag the Uploader emits status/metadata Records under.
+inline constexpr const char* FILE_STATUS_TAG = "dc.files";
+
+}  // namespace dc_bridge::uploader
+
+#endif  // DC_BRIDGE__UPLOADER__FILE_STATUS_TAG_HPP_

dc_bridge/include/dc_bridge/uploader/intent_queue.hpp

@@ -12,10 +12,11 @@
 // processing actually succeeds — no cap, no drop-oldest, no dead-letter. An upload
 // intent and its Files live and die together.
 //
-// The in-process wake-up channel (a condition_variable in BridgeNode) is still how the
-// worker learns "there's new work" cheaply; this directory is the crash-recovery state,
-// not the communication path — on startup every unacked intent left over from a
-// previous run is replayed, oldest-first, alongside live traffic.
+// This directory is the crash-recovery state, not a notification channel: on startup
+// every unacked intent left over from a previous run is replayed, oldest-first, alongside
+// live traffic. Since #446 split the Uploader into its own process, there is no
+// in-process wake-up signal at all — a Bridge process enqueues and a separate dc_uploader
+// process discovers new intents purely by polling rescan() (see that method below).
 #ifndef DC_BRIDGE__UPLOADER__INTENT_QUEUE_HPP_
 #define DC_BRIDGE__UPLOADER__INTENT_QUEUE_HPP_
 
@@ -93,6 +94,17 @@ class IntentQueue
   std::size_t size() const;
   bool empty() const;
 
+  /// Picks up any `*.json` file that exists on disk but isn't yet known to this
+  /// instance — the multi-process split (#446): a Bridge process enqueues intents while a
+  /// separate Uploader process holds the read side (next_ready()/ack()/record_failure()),
+  /// each with its own IntentQueue instance over the same directory. enqueue() only
+  /// updates its own caller's in-memory view, so the Uploader's instance never otherwise
+  /// learns about an intent a different process wrote; this is what closes that gap.
+  /// Already-known entries are left untouched (so in-flight backoff state survives);
+  /// returns the number of newly discovered intents. Safe to call from the same loop that
+  /// already polls for next_ready(), same cost as the constructor's initial scan.
+  std::size_t rescan();
+
 private:
   struct Entry
   {
@@ -105,6 +117,12 @@ class IntentQueue
   };
 
   std::string path_for(const std::string& id) const;
+  /// Every "*.json" filename currently on disk under `dir`, sorted oldest-first (ids sort
+  /// lexicographically in enqueue order — see intent_queue.cpp's make_id()).
+  static std::vector<std::string> list_json_names(const std::string& dir);
+  /// Parses one intent file's body into an Entry, or nullopt if it can't be read/parsed (a
+  /// "final" .json file should always be complete, since rename is atomic).
+  static std::optional<Entry> load_entry(const std::string& dir, const std::string& name);
 
   mutable std::mutex mutex_;
   std::string dir_;