feat(pj_base): attach_source_record runtime-host slot for the M3 source cache (0.28.0)

CHANGELOG.md

@@ -3,6 +3,37 @@
 All notable changes to `plotjuggler_sdk` are recorded here. Versioning policy is in
 [`CLAUDE.md`](./CLAUDE.md) → "Release Versioning".
 
+## [0.28.0]
+
+### Feature: source-record attachment for the host source cache (MINOR)
+
+`PJ_data_source_runtime_host_vtable_t` gains one tail slot,
+`attach_source_record(ctx, descriptor_json, out_error)`, with the C++ wrapper
+`DataSourceRuntimeHostView::attachSourceRecord`, also exposed on
+`DatasetIngestHostView`. At download start, a provider declares the canonical
+descriptor of the reproducible request its source answers. The host copies
+and stores the bytes verbatim, keys its cache on its own digest of those
+bytes, and scopes the record by the provider id from the binding, enabling
+the host-driven transparent source cache: captured downloads replayed from
+disk on the next restore of the same request, with no provider involvement on
+a hit.
+
+Contract: call on the stream thread; the last attachment before the first
+`push_message` on this ingest context wins. Byte-identical repeats before
+ingest are idempotent; any attachment after ingest begins is an error.
+The host may stage the record until its ingest transaction commits, so an
+in-place refill or replacing reload does not discard an early attachment.
+The host bounds and parses the descriptor and rejects unknown fields.
+Failure is a contract failure, never a trust verdict, and never affects
+ingest. Matching is byte-exact, so providers must serialize the same request
+identically. The descriptor is request identity, never parser policy, and
+must never carry authentication material. A host that predates the slot reads as
+"no caching" through `PJ_HAS_TAIL_SLOT`; the wrapper reports the absence
+explicitly so new plugins can detect it. Reachable from streaming sources and
+from toolbox parser-ingest contexts alike (both hold the runtime-host fat
+pointer). Runtime-host vtable size grows 104 → 112, `attach_source_record`
+at offset 104. ABI-appendable growth only; `abi/baseline.abi` untouched.
+
 ## [0.27.1]
 
 ### Fix: hosts validate a spliced `GridMap` right after attaching its bytes (PATCH)

VERSION

@@ -1 +1 @@
-0.27.1
+0.28.0

pj_base/CMakeLists.txt

@@ -173,6 +173,7 @@ if(PJ_BUILD_TESTS)
     tests/media_metadata_test.cpp
     tests/object_topic_metadata_test.cpp
     tests/push_message_test.cpp
+    tests/attach_source_record_test.cpp
     tests/notify_available_topics_test.cpp
     tests/dataset_ingest_view_test.cpp
     tests/descriptor_import_extension_test.cpp

pj_base/include/pj_base/data_source_protocol.h

@@ -341,6 +341,38 @@ typedef struct PJ_data_source_runtime_host_vtable_t {
    */
   bool (*notify_available_topics)(void* ctx, const PJ_available_topic_t* topics, uint64_t count, PJ_error_t* out_error)
       PJ_NOEXCEPT;
+
+  /**
+   * [stream-thread] Declare the reproducible request this source's data
+   * answers: the canonical descriptor JSON of the download (its "source
+   * record"), so the host can cache the ingested bytes and serve the next
+   * restore of the same request from disk. The host copies the bytes during
+   * the call and stores them VERBATIM; its cache is keyed on its own digest
+   * of those bytes — an internal keyspace, never required to agree with any
+   * provider identity scheme, and a layout round-trips the same bytes. What
+   * the plugin cannot spoof is the provider id: the host takes it from this
+   * binding and scopes the record with it.
+   *
+   * The host bounds and parses the descriptor and refuses anything it cannot
+   * fully account for (allowlist semantics: unknown fields are an error,
+   * never silently ignored; credential material never belongs in one). The
+   * descriptor is request identity, never parser policy — interpretation
+   * (timestamp fields, array limits) lives in the layout. Matching is
+   * byte-exact, so a provider re-serializing the same request must emit
+   * identical bytes.
+   *
+   * Call at download start: the last attach before the first push_message on
+   * this ingest context wins; an attach after ingest has begun is the error.
+   * The host may defer APPLYING the record until its ingest transaction
+   * commits (staged on the ingest context) — a committed in-place refill
+   * detaches records and a replacing reload gets a fresh context, so an
+   * early attach is not lost to either. Failure (malformed, over policy
+   * bounds, attach-after-ingest) returns false + error and never affects
+   * ingest — a contract failure, not a trust verdict (a refused record only
+   * means no caching). A host that predates this slot never caches; gate
+   * with PJ_HAS_TAIL_SLOT. Tail slot.
+   */
+  bool (*attach_source_record)(void* ctx, PJ_string_view_t descriptor_json, PJ_error_t* out_error) PJ_NOEXCEPT;
 } PJ_data_source_runtime_host_vtable_t;
 
 /** Fat pointer pairing a runtime host context with its vtable. */

pj_base/include/pj_base/sdk/data_source_host_views.hpp

@@ -198,6 +198,17 @@ class DataSourceRuntimeHostView {
   /// fall back to legacy behavior. Call on the poll/stream thread.
   [[nodiscard]] Status notifyAvailableTopics(Span<const AvailableTopic> topics) const;
 
+  /// Declare, on the stream thread at download start, the canonical descriptor of
+  /// the reproducible request this source answers, so the host can cache the
+  /// download (attach_source_record tail slot). The host copies and stores the
+  /// bytes verbatim and derives its cache key from them, scoped by the bound
+  /// provider id. The last attachment before this context's first pushMessage
+  /// wins; byte-identical repeats before ingest are idempotent. Attaching after
+  /// ingest begins or using a host that predates the slot returns an error
+  /// without affecting ingest. See PJ_data_source_runtime_host_vtable_t's
+  /// attach_source_record documentation for validation and commit semantics.
+  [[nodiscard]] Status attachSourceRecord(std::string_view descriptor_json) const;
+
   /// Push a message via a deferred FetchMessageData callable. The DataSource
   /// hands the host a callable that produces the payload bytes when invoked.
   /// The host applies the active ObjectIngestPolicy (resolved via the
@@ -427,6 +438,13 @@ class DatasetIngestHostView {
     return host_.pushMessage(handle, host_timestamp_ns, std::forward<FetchMessageData>(fetch_message_data));
   }
 
+  /// Declare this dataset's source record (the canonical descriptor of the
+  /// download) so the host can cache it. See
+  /// DataSourceRuntimeHostView::attachSourceRecord for the full contract.
+  [[nodiscard]] Status attachSourceRecord(std::string_view descriptor_json) const {
+    return host_.attachSourceRecord(descriptor_json);
+  }
+
   /// Narrow parser-only facade over the same underlying context.
   [[nodiscard]] ParserIngestHostView parserIngest() const noexcept {
     return host_.parserIngest();

pj_base/src/data_source_host_views.cpp

@@ -80,6 +80,20 @@ Status DataSourceRuntimeHostView::notifyAvailableTopics(Span<const AvailableTopi
   return okStatus();
 }
 
+Status DataSourceRuntimeHostView::attachSourceRecord(std::string_view descriptor_json) const {
+  if (!valid()) {
+    return unexpected(std::string("runtime host is not bound"));
+  }
+  if (!PJ_HAS_TAIL_SLOT(PJ_data_source_runtime_host_vtable_t, host_.vtable, attach_source_record)) {
+    return unexpected(std::string("runtime host does not expose attach_source_record"));
+  }
+  PJ_error_t err{};
+  if (!host_.vtable->attach_source_record(host_.ctx, sdk::toAbiString(descriptor_json), &err)) {
+    return unexpected(errorToString(err));
+  }
+  return okStatus();
+}
+
 MessageBoxButton DataSourceRuntimeHostView::showMessageBox(
     MessageBoxType type, std::string_view title, std::string_view message, int buttons) const {
   if (!valid() || host_.vtable->show_message_box == nullptr) {

pj_base/src/descriptor_import/provider_job.cpp

@@ -201,8 +201,10 @@ void JobControl::armWatchdog(std::chrono::milliseconds timeout, std::function<vo
     state_.watchdog_stop = false;
   }
   detail::JobState* state = &state_;
-  state_.watchdog = std::thread([state, timeout, on_expire = std::move(on_expire)]() {
+  std::binary_semaphore started{0};
+  state_.watchdog = std::thread([state, timeout, on_expire = std::move(on_expire), &started]() {
     std::unique_lock<std::mutex> lock(state->watchdog_mu);
+    started.release();
     const bool stopped = state->watchdog_cv.wait_for(lock, timeout, [state] { return state->watchdog_stop; });
     lock.unlock();
     if (!stopped) {
@@ -211,6 +213,7 @@ void JobControl::armWatchdog(std::chrono::milliseconds timeout, std::function<vo
       } catch (...) {}
     }
   });
+  started.acquire();
 }
 
 // ---------------------------------------------------------------------------

pj_base/tests/abi_layout_sentinels_test.cpp

@@ -246,7 +246,10 @@ static_assert(
     offsetof(PJ_data_source_runtime_host_vtable_t, notify_available_topics) == 96,
     "notify_available_topics tail slot pinned");
 static_assert(
-    sizeof(PJ_data_source_runtime_host_vtable_t) == 104, "Runtime host vtable size (update deliberately on append)");
+    offsetof(PJ_data_source_runtime_host_vtable_t, attach_source_record) == 104,
+    "attach_source_record tail slot pinned");
+static_assert(
+    sizeof(PJ_data_source_runtime_host_vtable_t) == 112, "Runtime host vtable size (update deliberately on append)");
 
 // --- Write-host vtables (ABI-APPENDABLE within v4) --------------------------
 static_assert(offsetof(PJ_source_write_host_vtable_t, abi_version) == 0, "source write host prefix pinned");

pj_base/tests/attach_source_record_test.cpp

@@ -0,0 +1,119 @@
+// Copyright 2026 Davide Faconti
+// SPDX-License-Identifier: Apache-2.0
+
+// Tests for the attach_source_record runtime-host tail slot:
+//
+//   1. DataSourceRuntimeHostView::attachSourceRecord flows the descriptor
+//      bytes through the slot; the host copies during the call.
+//   2. A host error (e.g. exceeding descriptor bounds) surfaces as the host's own
+//      message, never as a silent success.
+//   3. When the host predates the slot (short struct_size / NULL field), the
+//      call returns an explicit error — a NEW plugin on an OLD host detects
+//      "no caching" instead of degrading silently.
+
+#include <gtest/gtest.h>
+
+#include <cstddef>
+#include <cstdio>
+#include <string>
+
+#include "pj_base/data_source_protocol.h"
+#include "pj_base/sdk/data_source_host_views.hpp"
+
+namespace {
+
+// Mock runtime host — captures attach_source_record calls.
+class MockHost {
+ public:
+  MockHost() {
+    vtable_.protocol_version = 1;
+    vtable_.struct_size = sizeof(PJ_data_source_runtime_host_vtable_t);
+    vtable_.attach_source_record = &MockHost::attachThunk;
+    host_.ctx = this;
+    host_.vtable = &vtable_;
+  }
+
+  // Simulate an older host that predates the slot.
+  // Old host: shrink struct_size AND null the field.
+  void dropAttachSourceRecord() {
+    vtable_.attach_source_record = nullptr;
+    vtable_.struct_size = offsetof(PJ_data_source_runtime_host_vtable_t, attach_source_record);
+  }
+
+  // Host that reports a short struct_size but left a stale non-null pointer:
+  // the size gate alone must keep the slot unreachable.
+  void shrinkStructSizeOnly() {
+    vtable_.struct_size = offsetof(PJ_data_source_runtime_host_vtable_t, attach_source_record);
+  }
+
+  PJ::DataSourceRuntimeHostView view() const {
+    return PJ::DataSourceRuntimeHostView(host_);
+  }
+
+  std::string captured;
+  int call_count = 0;
+  bool refuse = false;
+
+ private:
+  static bool attachThunk(void* ctx, PJ_string_view_t descriptor_json, PJ_error_t* err) noexcept {
+    auto* self = static_cast<MockHost*>(ctx);
+    self->call_count++;
+    self->captured.assign(descriptor_json.data, descriptor_json.size);
+    if (self->refuse) {
+      if (err != nullptr) {
+        std::snprintf(err->message, sizeof(err->message), "descriptor exceeds host policy bounds");
+      }
+      return false;
+    }
+    return true;
+  }
+
+  PJ_data_source_runtime_host_vtable_t vtable_{};
+  PJ_data_source_runtime_host_t host_{};
+};
+
+TEST(AttachSourceRecordTest, DescriptorFlowsThroughSlot) {
+  MockHost host;
+  const std::string descriptor = R"({"kind":"example-request","v":1,"topics":["/a","/b"]})";
+
+  auto status = host.view().attachSourceRecord(descriptor);
+  ASSERT_TRUE(status) << (status ? "" : status.error());
+  EXPECT_EQ(host.call_count, 1);
+  EXPECT_EQ(host.captured, descriptor);
+}
+
+TEST(AttachSourceRecordTest, HostRefusalCarriesTheHostsReason) {
+  MockHost host;
+  host.refuse = true;
+
+  auto status = host.view().attachSourceRecord(R"({"v":1})");
+  ASSERT_FALSE(status);
+  EXPECT_NE(status.error().find("policy bounds"), std::string::npos);
+}
+
+TEST(AttachSourceRecordTest, ReturnsErrorWhenSlotMissing) {
+  MockHost host;
+  host.dropAttachSourceRecord();
+
+  auto status = host.view().attachSourceRecord(R"({"v":1})");
+  EXPECT_FALSE(status);  // explicit failure so a new plugin can fall back
+  EXPECT_EQ(host.call_count, 0);
+}
+
+TEST(AttachSourceRecordTest, ShortStructSizeAloneGatesTheSlot) {
+  MockHost host;
+  host.shrinkStructSizeOnly();  // stale non-null pointer past the reported size
+
+  auto status = host.view().attachSourceRecord(R"({"v":1})");
+  EXPECT_FALSE(status);
+  EXPECT_EQ(host.call_count, 0);
+}
+
+TEST(AttachSourceRecordTest, UnboundHostReportsNotBound) {
+  PJ::DataSourceRuntimeHostView view;  // default: no host
+  auto status = view.attachSourceRecord(R"({"v":1})");
+  ASSERT_FALSE(status);
+  EXPECT_NE(status.error().find("not bound"), std::string::npos);
+}
+
+}  // namespace

pj_plugins/docs/ARCHITECTURE.md

@@ -824,3 +824,21 @@ whole-source pause. Two independent additions, both `struct_size`/
 
 See `docs/data-source-guide.md` → "Per-topic pause (demand-driven
 subscription)" for the plugin-author walkthrough.
+
+## Source-cache attachment
+
+SDK 0.28 appends `attach_source_record(ctx, descriptor_json, out_error)` to
+the runtime-host vtable at offset 104, growing its size 104 → 112. Providers
+call it on the stream thread at download start. The host copies and stores
+the descriptor bytes verbatim, keys its cache on its own digest, and scopes
+the record by the provider id from the binding. Matching is byte-exact;
+the host bounds and parses the descriptor and rejects unknown fields.
+
+The last attachment before the ingest context's first `push_message` wins;
+byte-identical repeats before ingest are idempotent. Any attachment after
+ingest begins is an error. The host may stage the record until its ingest
+transaction commits so a refill or reload does not discard it. Failure never
+affects ingest. The descriptor identifies the request, carries no credentials,
+and leaves parser policy to the layout. `DataSourceRuntimeHostView` and
+`DatasetIngestHostView` expose `attachSourceRecord`; hosts predating the slot
+are detected with `PJ_HAS_TAIL_SLOT` and reported as an error (no caching).

pj_plugins/docs/data-source-guide.md

@@ -478,6 +478,7 @@ Access via `runtimeHost()`. Use this for lifecycle coordination and diagnostics.
 | `ensureParserBinding(request)` | Bind a parser for delegated ingest (see below). |
 | `pushMessage(handle, timestamp, fetch_message_data)` | Push a message through a parser binding via a deferred fetcher callable; the host invokes it per the active ObjectIngestPolicy (eager/lazy). |
 | `notifyAvailableTopics(topics)` | Advertise the full set of topics you *can* stream but have not subscribed, so the host lists and a-priori classifies them with no data flowing. See *Per-topic pause* below. |
+| `attachSourceRecord(descriptor_json)` | On the stream thread, declare the request descriptor for the host's source cache. The host stores the bytes verbatim and derives its cache key. Last attach before this context's first `pushMessage` wins; byte-identical repeats before ingest are idempotent. Errors after ingest begins or on hosts that predate the slot never affect ingest. See [Source-cache attachment](ARCHITECTURE.md#source-cache-attachment). |
 
 ## Optional Features
 

pj_plugins/tests/data_source_library_test.cpp

@@ -114,6 +114,7 @@ PJ_data_source_runtime_host_t makeRuntimeHost(bool with_encodings) {
       .list_available_encodings = rhListEncodings,
       .push_message = rhPushMessage,
       .notify_available_topics = nullptr,
+      .attach_source_record = nullptr,
   };
   static const PJ_data_source_runtime_host_vtable_t no_enc_vt = {
       .protocol_version = 1,
@@ -130,6 +131,7 @@ PJ_data_source_runtime_host_t makeRuntimeHost(bool with_encodings) {
       .list_available_encodings = nullptr,
       .push_message = rhPushMessage,
       .notify_available_topics = nullptr,
+      .attach_source_record = nullptr,
   };
   return PJ_data_source_runtime_host_t{
       .ctx = reinterpret_cast<void*>(0x2),

pj_plugins/tests/file_source_integration_test.cpp

@@ -155,6 +155,7 @@ PJ_data_source_runtime_host_t makeRuntimeHost(RuntimeHostState* state) {
       .list_available_encodings = nullptr,
       .push_message = nullptr,
       .notify_available_topics = nullptr,
+      .attach_source_record = nullptr,
   };
   return PJ_data_source_runtime_host_t{.ctx = state, .vtable = &vtable};
 }