From 0152611dd9a85196adbb7763f981b7c6c87613a5 Mon Sep 17 00:00:00 2001 From: Davide Faconti Date: Thu, 3 Sep 2026 22:45:32 +0200 Subject: [PATCH 1/4] feat(pj_base): attach_source_record runtime-host tail slot (0.28.0) One provider->host call, stated once at download start: the canonical descriptor JSON of the reproducible request a source answers. The host copies the bytes and derives the record identity itself (the plugin never supplies an identity, per the descriptor_import doctrine), enabling the host-driven transparent source cache. Byte-identical re-attach is idempotent; different bytes fail; failure is a contract failure, never a trust verdict, and never affects ingest. Old hosts read as no-caching via PJ_HAS_TAIL_SLOT; the C++ wrapper reports the absence explicitly. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 25 +++++ VERSION | 2 +- pj_base/CMakeLists.txt | 1 + .../include/pj_base/data_source_protocol.h | 26 +++++ .../pj_base/sdk/data_source_host_views.hpp | 8 ++ pj_base/src/data_source_host_views.cpp | 14 +++ pj_base/tests/abi_layout_sentinels_test.cpp | 5 +- pj_base/tests/attach_source_record_test.cpp | 96 +++++++++++++++++++ pj_plugins/tests/data_source_library_test.cpp | 2 + .../tests/file_source_integration_test.cpp | 1 + 10 files changed, 178 insertions(+), 2 deletions(-) create mode 100644 pj_base/tests/attach_source_record_test.cpp diff --git a/CHANGELOG.md b/CHANGELOG.md index 18aa7233..da6bda12 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,31 @@ 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`. A provider states once, at +download start, the canonical descriptor of the reproducible request its +source answers; the host copies the bytes, canonicalizes them under its +descriptor policy, and derives the record identity itself (the plugin never +supplies an identity — the `descriptor_import_protocol.h` doctrine), 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: thread-safe; bytes copied during the call; at most one record per +source (byte-identical repeat = idempotent success, different bytes = error); +failure is a contract failure, never a trust verdict, and never affects +ingest. 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). ABI-appendable growth only; `abi/baseline.abi` untouched. + ## [0.27.0] ### Feature: shared timestamp arithmetic and axis policy (MINOR) diff --git a/VERSION b/VERSION index 1b58cc10..697f087f 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.27.0 +0.28.0 diff --git a/pj_base/CMakeLists.txt b/pj_base/CMakeLists.txt index f35ae922..7b940edd 100644 --- a/pj_base/CMakeLists.txt +++ b/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 diff --git a/pj_base/include/pj_base/data_source_protocol.h b/pj_base/include/pj_base/data_source_protocol.h index c9b773f0..a1bc7fae 100644 --- a/pj_base/include/pj_base/data_source_protocol.h +++ b/pj_base/include/pj_base/data_source_protocol.h @@ -341,6 +341,32 @@ 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; + + /** + * [thread-safe] Declare, once per source, 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, canonicalizes them under its descriptor policy and + * derives the record identity ITSELF, scoped by this source's provider — + * the plugin never supplies an identity, so it cannot alias one request's + * artifact to another's (same doctrine as descriptor_import_protocol.h). + * + * The descriptor is request identity, never parser policy: interpretation + * (timestamp fields, array limits) lives in the layout. It must never carry + * authentication material; the host's allowlist rejects credential-shaped + * fields (api_key, cert_path, allow_insecure) and refuses the record. + * + * Call at download start, before the first push_message; ingest that + * happened before the call may not be captured. At most one record per + * source: repeating the call with byte-identical JSON is an idempotent + * success, different bytes fail. Failure (malformed, over policy bounds, + * conflicting re-attach) returns false + error and never affects ingest — + * a contract failure, not a trust verdict (trust is decided by the host at + * import time, and 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. */ diff --git a/pj_base/include/pj_base/sdk/data_source_host_views.hpp b/pj_base/include/pj_base/sdk/data_source_host_views.hpp index a40ec0a5..2c26d5fb 100644 --- a/pj_base/include/pj_base/sdk/data_source_host_views.hpp +++ b/pj_base/include/pj_base/sdk/data_source_host_views.hpp @@ -198,6 +198,14 @@ class DataSourceRuntimeHostView { /// fall back to legacy behavior. Call on the poll/stream thread. [[nodiscard]] Status notifyAvailableTopics(Span topics) const; + /// Declare, once per source 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 the bytes and + /// derives the record identity itself. Errors when the host predates the + /// slot — a NEW plugin on an OLD host can detect the absence (no caching) + /// instead of degrading silently. + [[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 diff --git a/pj_base/src/data_source_host_views.cpp b/pj_base/src/data_source_host_views.cpp index 1fa73a28..62f27327 100644 --- a/pj_base/src/data_source_host_views.cpp +++ b/pj_base/src/data_source_host_views.cpp @@ -80,6 +80,20 @@ Status DataSourceRuntimeHostView::notifyAvailableTopics(Spanattach_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) { diff --git a/pj_base/tests/abi_layout_sentinels_test.cpp b/pj_base/tests/abi_layout_sentinels_test.cpp index 77da1820..6c3988c7 100644 --- a/pj_base/tests/abi_layout_sentinels_test.cpp +++ b/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"); diff --git a/pj_base/tests/attach_source_record_test.cpp b/pj_base/tests/attach_source_record_test.cpp new file mode 100644 index 00000000..3681f406 --- /dev/null +++ b/pj_base/tests/attach_source_record_test.cpp @@ -0,0 +1,96 @@ +// 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. conflicting re-attach) 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 + +#include +#include +#include + +#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. + void dropAttachSourceRecord() { + vtable_.attach_source_record = nullptr; + 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(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), "source already carries a different record"); + } + 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":"mosaico-sequence","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("different record"), 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); +} + +} // namespace diff --git a/pj_plugins/tests/data_source_library_test.cpp b/pj_plugins/tests/data_source_library_test.cpp index 5503ea54..d357a0c7 100644 --- a/pj_plugins/tests/data_source_library_test.cpp +++ b/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(0x2), diff --git a/pj_plugins/tests/file_source_integration_test.cpp b/pj_plugins/tests/file_source_integration_test.cpp index 220a2afc..26c52109 100644 --- a/pj_plugins/tests/file_source_integration_test.cpp +++ b/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}; } From cc4bcc930102583866c3877df6ef7eb7f4e32dc0 Mon Sep 17 00:00:00 2001 From: Davide Faconti Date: Thu, 3 Sep 2026 23:01:17 +0200 Subject: [PATCH 2/4] docs(pj_base): source-record contract fixed to what a host can implement MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The record is stored verbatim and keyed on the host's own digest (an internal keyspace — never required to agree with a provider identity scheme); the unspoofable datum is the provider id, taken from the binding. Refusal is allowlist-semantics (unknown fields are an error), not a field denylist. Last attach before the first push_message wins and the host may stage the record until its ingest transaction commits, so a replacing reload or in-place refill never silently loses it. Thread tag matches push_message ([stream-thread]). DatasetIngestHostView gains the forward; guide/architecture docs list the slot; tests add the unbound-host and short-struct_size-with-stale-pointer gates. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 2 +- .../include/pj_base/data_source_protocol.h | 46 +++++++++++-------- .../pj_base/sdk/data_source_host_views.hpp | 7 +++ pj_base/tests/attach_source_record_test.cpp | 27 ++++++++++- pj_plugins/docs/ARCHITECTURE.md | 6 ++- pj_plugins/docs/data-source-guide.md | 1 + 6 files changed, 65 insertions(+), 24 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index da6bda12..b71a00e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,7 +26,7 @@ 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). ABI-appendable growth only; `abi/baseline.abi` untouched. +pointer). Runtime-host vtable size grows 104 → 112, `attach_source_record` at offset 104. ABI-appendable growth only; `abi/baseline.abi` untouched. ## [0.27.0] diff --git a/pj_base/include/pj_base/data_source_protocol.h b/pj_base/include/pj_base/data_source_protocol.h index a1bc7fae..e8ac2610 100644 --- a/pj_base/include/pj_base/data_source_protocol.h +++ b/pj_base/include/pj_base/data_source_protocol.h @@ -343,28 +343,34 @@ typedef struct PJ_data_source_runtime_host_vtable_t { PJ_NOEXCEPT; /** - * [thread-safe] Declare, once per source, 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, canonicalizes them under its descriptor policy and - * derives the record identity ITSELF, scoped by this source's provider — - * the plugin never supplies an identity, so it cannot alias one request's - * artifact to another's (same doctrine as descriptor_import_protocol.h). + * [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 descriptor is request identity, never parser policy: interpretation - * (timestamp fields, array limits) lives in the layout. It must never carry - * authentication material; the host's allowlist rejects credential-shaped - * fields (api_key, cert_path, allow_insecure) and refuses the record. + * 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, before the first push_message; ingest that - * happened before the call may not be captured. At most one record per - * source: repeating the call with byte-identical JSON is an idempotent - * success, different bytes fail. Failure (malformed, over policy bounds, - * conflicting re-attach) returns false + error and never affects ingest — - * a contract failure, not a trust verdict (trust is decided by the host at - * import time, and a refused record only means no caching). A host that - * predates this slot never caches; gate with PJ_HAS_TAIL_SLOT. Tail slot. + * 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; diff --git a/pj_base/include/pj_base/sdk/data_source_host_views.hpp b/pj_base/include/pj_base/sdk/data_source_host_views.hpp index 2c26d5fb..1f65794f 100644 --- a/pj_base/include/pj_base/sdk/data_source_host_views.hpp +++ b/pj_base/include/pj_base/sdk/data_source_host_views.hpp @@ -435,6 +435,13 @@ class DatasetIngestHostView { return host_.pushMessage(handle, host_timestamp_ns, std::forward(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(); diff --git a/pj_base/tests/attach_source_record_test.cpp b/pj_base/tests/attach_source_record_test.cpp index 3681f406..1996d64a 100644 --- a/pj_base/tests/attach_source_record_test.cpp +++ b/pj_base/tests/attach_source_record_test.cpp @@ -14,7 +14,7 @@ #include #include -#include +#include #include #include "pj_base/data_source_protocol.h" @@ -34,11 +34,18 @@ class MockHost { } // 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_); } @@ -67,7 +74,7 @@ class MockHost { TEST(AttachSourceRecordTest, DescriptorFlowsThroughSlot) { MockHost host; - const std::string descriptor = R"({"kind":"mosaico-sequence","v":1,"topics":["/a","/b"]})"; + 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()); @@ -93,4 +100,20 @@ TEST(AttachSourceRecordTest, ReturnsErrorWhenSlotMissing) { 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 diff --git a/pj_plugins/docs/ARCHITECTURE.md b/pj_plugins/docs/ARCHITECTURE.md index 0aa335e5..67eda20a 100644 --- a/pj_plugins/docs/ARCHITECTURE.md +++ b/pj_plugins/docs/ARCHITECTURE.md @@ -803,7 +803,11 @@ whole-source pause. Two independent additions, both `struct_size`/ - **Plugin → host advertise.** A second tail slot on the runtime host, `notify_available_topics(ctx, topics, count, out_error)` (offset 96, - growing `sizeof(PJ_data_source_runtime_host_vtable_t)` 96 → 104), carrying + growing `sizeof(PJ_data_source_runtime_host_vtable_t) — + followed in 0.28 by `attach_source_record` (offset 104, size 104 → 112, + the source-cache record declaration)` 96 → 104) — + followed in 0.28 by `attach_source_record` (offset 104, size 104 → 112, + the source-cache record declaration), carrying `PJ_available_topic_t{topic_name, parser_encoding, type_name, schema}` — the same parser-identifying fields as `PJ_parser_binding_request_t` minus `parser_config_json` (not yet known pre-subscription), so the host can diff --git a/pj_plugins/docs/data-source-guide.md b/pj_plugins/docs/data-source-guide.md index b0f31825..abe3b096 100644 --- a/pj_plugins/docs/data-source-guide.md +++ b/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)` | Declare the canonical descriptor of the reproducible request this source answers, so the host can cache the download and restore it from disk next time. Last attach before the first `pushMessage` wins; errors on hosts that predate the slot (no caching). | ## Optional Features From be9d94fdade4f7c9d48ac6e03c7b2d8ba36ef90c Mon Sep 17 00:00:00 2001 From: Davide Faconti Date: Thu, 3 Sep 2026 23:47:17 +0200 Subject: [PATCH 3/4] fix(pj_base): wait until watchdog is armed --- pj_base/src/descriptor_import/provider_job.cpp | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/pj_base/src/descriptor_import/provider_job.cpp b/pj_base/src/descriptor_import/provider_job.cpp index 8bfb9105..1f22ee22 100644 --- a/pj_base/src/descriptor_import/provider_job.cpp +++ b/pj_base/src/descriptor_import/provider_job.cpp @@ -201,8 +201,10 @@ void JobControl::armWatchdog(std::chrono::milliseconds timeout, std::function 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 Date: Sat, 5 Sep 2026 09:32:47 +0200 Subject: [PATCH 4/4] docs(pj_base): align source-record attachment contract --- CHANGELOG.md | 28 +++++++++++-------- .../pj_base/sdk/data_source_host_views.hpp | 13 +++++---- pj_base/tests/attach_source_record_test.cpp | 6 ++-- pj_plugins/docs/ARCHITECTURE.md | 24 ++++++++++++---- pj_plugins/docs/data-source-guide.md | 2 +- 5 files changed, 48 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 92422e25..1438cc03 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,24 +9,30 @@ All notable changes to `plotjuggler_sdk` are recorded here. Versioning policy is `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`. A provider states once, at -download start, the canonical descriptor of the reproducible request its -source answers; the host copies the bytes, canonicalizes them under its -descriptor policy, and derives the record identity itself (the plugin never -supplies an identity — the `descriptor_import_protocol.h` doctrine), enabling +`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: thread-safe; bytes copied during the call; at most one record per -source (byte-identical repeat = idempotent success, different bytes = error); -failure is a contract failure, never a trust verdict, and never affects -ingest. The descriptor is request identity, never parser policy, and must -never carry authentication material. A host that predates the slot reads as +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. +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] diff --git a/pj_base/include/pj_base/sdk/data_source_host_views.hpp b/pj_base/include/pj_base/sdk/data_source_host_views.hpp index 1f65794f..2fff04db 100644 --- a/pj_base/include/pj_base/sdk/data_source_host_views.hpp +++ b/pj_base/include/pj_base/sdk/data_source_host_views.hpp @@ -198,12 +198,15 @@ class DataSourceRuntimeHostView { /// fall back to legacy behavior. Call on the poll/stream thread. [[nodiscard]] Status notifyAvailableTopics(Span topics) const; - /// Declare, once per source at download start, the canonical descriptor of + /// 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 the bytes and - /// derives the record identity itself. Errors when the host predates the - /// slot — a NEW plugin on an OLD host can detect the absence (no caching) - /// instead of degrading silently. + /// 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 diff --git a/pj_base/tests/attach_source_record_test.cpp b/pj_base/tests/attach_source_record_test.cpp index 1996d64a..e97fafa0 100644 --- a/pj_base/tests/attach_source_record_test.cpp +++ b/pj_base/tests/attach_source_record_test.cpp @@ -5,7 +5,7 @@ // // 1. DataSourceRuntimeHostView::attachSourceRecord flows the descriptor // bytes through the slot; the host copies during the call. -// 2. A host error (e.g. conflicting re-attach) surfaces as the host's own +// 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 @@ -61,7 +61,7 @@ class MockHost { self->captured.assign(descriptor_json.data, descriptor_json.size); if (self->refuse) { if (err != nullptr) { - std::snprintf(err->message, sizeof(err->message), "source already carries a different record"); + std::snprintf(err->message, sizeof(err->message), "descriptor exceeds host policy bounds"); } return false; } @@ -88,7 +88,7 @@ TEST(AttachSourceRecordTest, HostRefusalCarriesTheHostsReason) { auto status = host.view().attachSourceRecord(R"({"v":1})"); ASSERT_FALSE(status); - EXPECT_NE(status.error().find("different record"), std::string::npos); + EXPECT_NE(status.error().find("policy bounds"), std::string::npos); } TEST(AttachSourceRecordTest, ReturnsErrorWhenSlotMissing) { diff --git a/pj_plugins/docs/ARCHITECTURE.md b/pj_plugins/docs/ARCHITECTURE.md index 67eda20a..5393a2be 100644 --- a/pj_plugins/docs/ARCHITECTURE.md +++ b/pj_plugins/docs/ARCHITECTURE.md @@ -803,11 +803,7 @@ whole-source pause. Two independent additions, both `struct_size`/ - **Plugin → host advertise.** A second tail slot on the runtime host, `notify_available_topics(ctx, topics, count, out_error)` (offset 96, - growing `sizeof(PJ_data_source_runtime_host_vtable_t) — - followed in 0.28 by `attach_source_record` (offset 104, size 104 → 112, - the source-cache record declaration)` 96 → 104) — - followed in 0.28 by `attach_source_record` (offset 104, size 104 → 112, - the source-cache record declaration), carrying + growing `sizeof(PJ_data_source_runtime_host_vtable_t)` 96 → 104), carrying `PJ_available_topic_t{topic_name, parser_encoding, type_name, schema}` — the same parser-identifying fields as `PJ_parser_binding_request_t` minus `parser_config_json` (not yet known pre-subscription), so the host can @@ -828,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). diff --git a/pj_plugins/docs/data-source-guide.md b/pj_plugins/docs/data-source-guide.md index abe3b096..70166d00 100644 --- a/pj_plugins/docs/data-source-guide.md +++ b/pj_plugins/docs/data-source-guide.md @@ -478,7 +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)` | Declare the canonical descriptor of the reproducible request this source answers, so the host can cache the download and restore it from disk next time. Last attach before the first `pushMessage` wins; errors on hosts that predate the slot (no caching). | +| `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