| 1677 | } |
| 1678 | |
| 1679 | bool sourceObjectPushLazy( |
| 1680 | void* ctx, PJ_object_topic_handle_t topic, int64_t timestamp_ns, PJ_lazy_fetch_fn_t fetch_fn, void* fetch_ctx, |
| 1681 | void (*fetch_ctx_destroy)(void*), PJ_error_t* out_error) noexcept { |
| 1682 | auto* impl = static_cast<DatastoreSourceObjectWriteHostState*>(ctx); |
| 1683 | if (fetch_fn == nullptr) { |
| 1684 | if (fetch_ctx_destroy != nullptr) { |
| 1685 | fetch_ctx_destroy(fetch_ctx); |
| 1686 | } |
| 1687 | propagateError(out_error, "fetch_fn must not be null"); |
| 1688 | return false; |
| 1689 | } |
| 1690 | auto* target = impl->target.load(std::memory_order_acquire); |
| 1691 | try { |
| 1692 | // shared_ptr keeps the ctx holder alive as long as ObjectStore keeps |
| 1693 | // the lambda; destructor runs exactly once when ObjectStore drops the |
| 1694 | // entry (retention, evict, removeTopic, clear, or store teardown). |
| 1695 | auto holder = std::make_shared<PluginFetchCtx>(fetch_fn, fetch_ctx, fetch_ctx_destroy); |
| 1696 | // Plugins return raw bytes via the C ABI; wrap them as a PayloadView whose |
| 1697 | // anchor is a shared_ptr<const vector<uint8_t>>, per the pushLazy contract. |
| 1698 | // Target pointer comes from the atomic swap layer (so writes follow the |
| 1699 | // current target store, not a captured-at-construction one). |
| 1700 | auto closure = [holder]() -> sdk::PayloadView { return sdk::makePayloadView(holder->invoke()); }; |
| 1701 | auto result = target->pushLazy(ObjectTopicId{topic.id}, timestamp_ns, std::move(closure)); |
| 1702 | if (!result) { |
| 1703 | impl->setError(result.error()); |
| 1704 | propagateError(out_error, impl->last_error.c_str()); |
| 1705 | // `holder` is the only reference to the ctx on failure; dropping it |
| 1706 | // runs fetch_ctx_destroy exactly once (the destructor already does it). |
| 1707 | return false; |
| 1708 | } |
| 1709 | impl->last_error.clear(); |
| 1710 | return true; |
| 1711 | } catch (const std::exception& e) { |
| 1712 | impl->setError(e.what()); |
| 1713 | propagateError(out_error, impl->last_error.c_str()); |
| 1714 | // On exception before the ObjectStore took ownership, PluginFetchCtx's |
| 1715 | // destructor runs as part of shared_ptr teardown — single destroy call. |
| 1716 | return false; |
| 1717 | } catch (...) { |
| 1718 | impl->setError("pushLazy: unknown exception"); |
| 1719 | propagateError(out_error, impl->last_error.c_str()); |
| 1720 | return false; |
| 1721 | } |
| 1722 | } |
| 1723 | |
| 1724 | void sourceObjectSetRetentionBudget( |
| 1725 | void* ctx, PJ_object_topic_handle_t topic, int64_t time_window_ns, uint64_t max_memory_bytes) noexcept { |