MINIFICPP-2845 - Monotonic provenance event identifiers, iterations support, refactor

[adamdebreceni]

Depends on #2195

Thank you for submitting a contribution to Apache NiFi - MiNiFi C++.

In order to streamline the review of the contribution we ask you to ensure the following steps have been taken:

For all changes:

• Is there a JIRA ticket associated with this PR? Is it referenced in the commit message?

• Does your PR title start with MINIFICPP-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.

• Has your PR been rebased against the latest commit within the target branch (typically main)?

• Is your initial contribution a single, squashed commit?

For code changes:

• If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under ASF 2.0?
• If applicable, have you updated the LICENSE file?
• If applicable, have you updated the NOTICE file?

For documentation related changes:

• Have you ensured that format looks appropriate for the output in which it is rendered?

Note:

Please ensure that once the PR is submitted, you check GitHub Actions CI results for build issues and submit an update to your PR as soon as possible.

[szaszm]

I believe the "repo" naming of identifiers that have to be provenance repo was a mistake. I'm glad the type makes it clearer now, but I'd change the parameter and member names too. provenance_repo or prov_repo seem like good options to me.

[szaszm]

why do we need these UUID increments? I believe the Identified class is not meant to represent integer auto-increment keys.

[adamdebreceni]

currently the RocksDbProvenanceRepository uses the event_uuid for key, we could either bring back the retired SequentialIdGenerator, we could replace the increment operators with a next() method to not cause confusion, or we could use an uint64_t (event ordinal) as key, which one do you prefer?

[Copilot]

Pull request overview

Introduces a dedicated provenance repository API with monotonic event identifiers and cursor-based reporting.

Changes:

• Adds provenance-specific append and iteration APIs.
• Implements persistent monotonic identifiers in RocksDB.
• Refactors reporting, runtime wiring, and tests to use the new API.

Reviewed changes

Copilot reviewed 43 out of 43 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
minifi-api/include/minifi-cpp/provenance/ProvenanceRepository.h	Adds provenance repository interface.
minifi-api/include/minifi-cpp/provenance/Provenance.h	Adds event ordinals and vector-backed reporting.
minifi-api/include/minifi-cpp/core/Repository.h	Removes generic provenance operations.
minifi-api/include/minifi-cpp/core/ProcessContext.h	Returns typed provenance repositories.
minifi-api/common/include/minifi-cpp/utils/Id.h	Declares identifier increment operators.
minifi_main/MiNiFiMain.cpp	Casts configured repository to the provenance API.
libminifi/test/unit/SchedulingAgentTests.cpp	Updates test repository typing.
libminifi/test/persistence-tests/PersistenceTests.cpp	Updates persistence fixtures.
libminifi/test/libtest/unit/TestControllerWithFlow.cpp	Updates test flow setup.
libminifi/test/libtest/unit/TestBase.h	Updates test plan provenance types.
libminifi/test/libtest/unit/TestBase.cpp	Implements updated test plan API.
libminifi/test/libtest/unit/ProvenanceTestHelper.h	Implements test provenance repository operations.
libminifi/test/libtest/integration/IntegrationBase.cpp	Updates integration repository setup.
libminifi/test/integration/ControllerServiceIntegrationTests.cpp	Updates repository inference.
libminifi/test/integration/C2PauseResumeTest.cpp	Updates repository inference.
libminifi/test/flow-tests/SessionTests.cpp	Casts the no-op repository.
libminifi/src/provenance/Provenance.cpp	Uses batched provenance appends.
libminifi/src/FlowController.cpp	Accepts the typed repository.
libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp	Adds cursor-based reporting and ordinals.
libminifi/src/core/ProcessContextImpl.cpp	Stores the typed repository.
libminifi/src/core/flow/StructuredConfiguration.cpp	Assigns reporting-task UUIDs.
libminifi/include/TimerDrivenSchedulingAgent.h	Updates repository type.
libminifi/include/ThreadedSchedulingAgent.h	Updates repository type.
libminifi/include/SchedulingAgent.h	Stores the typed repository.
libminifi/include/provenance/Provenance.h	Implements ordinals and vector-backed events.
libminifi/include/FlowController.h	Exposes the typed repository.
libminifi/include/EventDrivenSchedulingAgent.h	Updates repository type.
libminifi/include/CronDrivenSchedulingAgent.h	Updates repository type.
libminifi/include/core/repository/VolatileProvenanceRepository.h	Implements volatile event appends.
libminifi/include/core/repository/NoOpThreadedRepository.h	Implements no-op provenance operations.
libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h	Uses typed event records.
libminifi/include/core/ProcessContextImpl.h	Updates context repository types.
extensions/standard-processors/tests/unit/ProcessorTests.cpp	Adapts reporting tests to getEvents.
extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h	Updates test repository inference.
extensions/rocksdb-repos/tests/RepoTests.cpp	Uses renamed RocksDB repository.
extensions/rocksdb-repos/tests/ProvenanceTests.cpp	Tests the append API.
extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp	Adds cursor and monotonicity tests.
extensions/rocksdb-repos/RocksDbProvenanceRepository.h	Defines the renamed RocksDB implementation.
extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp	Implements counters, cursors, and iteration.
extensions/rocksdb-repos/ProvenanceRepository.cpp	Removes the previous implementation.
core-framework/src/core/Repository.cpp	Removes generic element storage.
core-framework/include/core/Repository.h	Removes generic provenance defaults.
core-framework/common/src/utils/Id.cpp	Implements identifier incrementing.

Suppressed comments (4)

extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp:131

• A persisted cursor is expected to be parsed so the reporting task can fall back when it is invalid, but every string is currently accepted. A malformed/high-valued state such as zzzz will seek past every UUID and make provenance reporting return no records forever; validate it as an Identifier and return nullptr on failure.

std::unique_ptr<ProvenanceRepository::Cursor> RocksDbProvenanceRepository::cursorFromString(std::optional<std::string> cursor_str) {
  if (cursor_str.has_value()) {
    return std::make_unique<EventCursor>(cursor_str.value());
  }
  return std::make_unique<EventCursor>("");
}

extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp:174

• The iterator status is never checked after traversal, so a RocksDB read error is returned as a successful partial result and the reporting task can persist its cursor. Check it->status() before updating the cursor and return std::unexpected on failure, consistent with RocksDbStateStorage iteration.

  if (event_cursor) {
    event_cursor->event_id_ = last_event_id;
  }
  return records;

libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp:224

• A false return from transmitPayload only yields and then reaches this cursor update, marking records as consumed even though they were not transmitted. Return from the trigger on the false path, just as the exception path does, so the same cursor batch is retried.

  if (cursor) {
    // no need to delete just update the state
    std::unordered_map<std::string, std::string> state_map;
    state_map["cursor"] = cursor->toString();
    if (!state_manager->set(state_map)) {

extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp:169

• Failed deserialization is silently skipped while last_event_id still advances. Once a later valid record is returned, the reporting task persists that advanced cursor, permanently losing the malformed record from the reporting stream. Return an error instead of advancing past an event that could not be decoded.

    if (eventRead->deserialize(stream)) {
      records.push_back(eventRead);
      if (--max_size == 0) {
        break;
      }
    }

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.