diff --git a/CHANGELOG.md b/CHANGELOG.md index cff1bc71..1438cc03 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/VERSION b/VERSION index 83b47304..697f087f 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.27.1 +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..e8ac2610 100644 --- a/pj_base/include/pj_base/data_source_protocol.h +++ b/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. */ 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..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,6 +198,17 @@ class DataSourceRuntimeHostView { /// fall back to legacy behavior. Call on the poll/stream thread. [[nodiscard]] Status notifyAvailableTopics(Span 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(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/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/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 + +#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. + // 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(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 diff --git a/pj_plugins/docs/ARCHITECTURE.md b/pj_plugins/docs/ARCHITECTURE.md index 0aa335e5..5393a2be 100644 --- a/pj_plugins/docs/ARCHITECTURE.md +++ b/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). diff --git a/pj_plugins/docs/data-source-guide.md b/pj_plugins/docs/data-source-guide.md index b0f31825..70166d00 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)` | 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 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}; }