registry_projector.hpp

Go to the documentation of this file.

#pragma once
 
#include <cstddef>
#include <cstdint>
#include <string_view>
 
#include "native.h"
#include <asio.hpp>
 
#include "event_projector.hpp"
#include "sqlite_hot_store.hpp"
#include "registry.hpp"
 
namespace dcn::registry
{
    class RegistryProjector final : public events::IEventProjector
    {
        public:
            RegistryProjector(
                events::SQLiteHotStore & store,
                Registry & registry,
                const asio::strand<asio::io_context::executor_type> & write_strand,
                events::ProjectorRetryConfig retry = {});
 
            std::string_view id() const override;
 
            std::int64_t cursor() const override { return _cursor; }
 
            asio::awaitable<std::size_t> projectBatch(std::size_t limit, std::int64_t now_ms) override;
 
        private:
            // Returns true  → row materialized (or already present); advance past it.
            // Returns false → materialization failed; projectBatch retries the row on
            //                 later passes and dead-letters it after
            //                 _retry.max_attempts consecutive failures.
            asio::awaitable<bool> _materializeOne(const events::ChangeRecord & row);
 
            // Retries dead-lettered rows (throttled by _retry.sweep_interval_ms)
            // and clears their bit on success. Returns the number of rows resolved.
            asio::awaitable<std::size_t> _sweepDeadLetters(std::size_t limit, std::int64_t now_ms);
 
            events::SQLiteHotStore & _store;
            Registry & _registry;
            asio::strand<asio::io_context::executor_type> _write_strand;
            events::ProjectorRetryConfig _retry;
            std::int64_t _cursor{0};
 
            // Consecutive-failure tracking for the row currently at the cursor head.
            std::int64_t _failing_seq{0};
            std::size_t _failure_count{0};
 
            std::int64_t _last_sweep_ms{0};
    };
}