arachne::coordination::operational_ledger Class Reference

#include <include/arachne/coordinator.hpp>

Collaboration diagram for arachne::coordination::operational_ledger:

Classes
struct	implementation

Public Member Functions
	operational_ledger (std::filesystem::path database_path, std::optional< std::filesystem::path > legacy_inbox_root=std::nullopt)
	~operational_ledger ()
	operational_ledger (const operational_ledger &)=delete
operational_ledger &	operator= (const operational_ledger &)=delete
	operational_ledger (operational_ledger &&)=delete
operational_ledger &	operator= (operational_ledger &&)=delete
envelope_record	intake (const intake_request &request)
envelope_record	transition (std::string_view envelope_id, cocoon_status next, std::string_view actor_ref, std::string_view reason={})
envelope_record	get (std::string_view envelope_id) const
std::vector< envelope_record >	list (cocoon_status status) const
std::vector< state_event >	history (std::string_view envelope_id) const
void	capture_inbox_baseline (const std::filesystem::path &inbox_root)
std::vector< verification_issue >	verify_inbox (const std::filesystem::path &inbox_root) const
bool	retire_queued_payload (std::string_view envelope_id, const std::filesystem::path &internal_queue_root, std::optional< std::filesystem::path > legacy_inbox_root=std::nullopt)
accumulation_state	accumulation () const
bool	should_integrate (const accumulation_policy &policy) const
bool	claim_logical_run (std::string_view run_id, std::string_view graph_domain, std::string_view logical_date, std::string_view configuration_sha256, bool retry_failed=false, bool resume_running=false)
void	bind_product_run_inputs (std::string_view run_id, const std::vector< std::string > &envelope_ids)
std::vector< envelope_record >	product_run_inputs (std::string_view run_id) const
void	finish_integrated_product_run (std::string_view run_id, std::string_view manifest_ref)
void	finish_run (std::string_view run_id, std::string_view status, std::string_view manifest_ref={})
const std::filesystem::path &	path () const noexcept

Private Attributes
implementation *	impl_
std::filesystem::path	path_
std::optional< std::filesystem::path >	legacy_inbox_root_

Detailed Description

Definition at line 81 of file coordinator.hpp.

Constructor & Destructor Documentation

operational_ledger() [1/3]

arachne::coordination::operational_ledger::operational_ledger ( std::filesystem::path database_path, std::optional< std::filesystem::path > legacy_inbox_root = std::nullopt )

explicit

Definition at line 564 of file coordinator.cpp.

    : impl_(nullptr) {
    reject_symlink_components(database_path, "operational ledger path");
    path_ = weakly_canonical_or_absolute(database_path);
    if (legacy_inbox_root) {
        reject_symlink_components(*legacy_inbox_root, "legacy inbox path");
        legacy_inbox_root_ = weakly_canonical_or_absolute(*legacy_inbox_root);
        if (path_is_within(path_, *legacy_inbox_root_)) {
            throw std::invalid_argument(
                "operational ledger must remain outside the legacy inbox"
            );
        }
    }
    if (path_.has_parent_path()) {
        std::filesystem::create_directories(path_.parent_path());
    }
    impl_ = new implementation(path_);
}

References impl_.

~operational_ledger()

arachne::coordination::operational_ledger::~operational_ledger

(

)

Definition at line 586 of file coordinator.cpp.

{ delete impl_; }

References impl_.

operational_ledger() [2/3]

arachne::coordination::operational_ledger::operational_ledger ( const operational_ledger & )

delete

operational_ledger() [3/3]

arachne::coordination::operational_ledger::operational_ledger ( operational_ledger && )

delete

Member Function Documentation

accumulation()

accumulation_state arachne::coordination::operational_ledger::accumulation ( ) const

nodiscard

Definition at line 1083 of file coordinator.cpp.

                                                      {
        return false;
    }
    if (error || std::filesystem::is_symlink(status)
        || !std::filesystem::is_regular_file(status)) {
        throw std::runtime_error(
            "queued payload is not a removable regular file"
        );
    }
    const auto retained = fingerprint_regular_file(
        envelope.payload_ref, std::numeric_limits<std::uintmax_t>::max(),
        "queued payload"
    );
    if (retained.sha256 != envelope.payload_sha256
        || retained.byte_length != envelope.byte_length) {
        throw std::runtime_error(

bind_product_run_inputs()

void arachne::coordination::operational_ledger::bind_product_run_inputs	(	std::string_view	run_id,
		const std::vector< std::string > &	envelope_ids )

Definition at line 1241 of file coordinator.cpp.

                                      {
                date_was_claimed = true;
                prior_configuration = column_text(prior_date.get(), 0);
                const std::string status = column_text(prior_date.get(), 1);
                if (status == "running" || status == "succeeded") {
                    execute(database, "COMMIT;");
                    return false;
                }
            } else if (result != SQLITE_DONE) {
                throw_sqlite(database, "inspect logical-date guard");
            }
        }
        if (date_was_claimed) {
            if (prior_configuration != configuration_sha256) {
                throw std::logic_error(
                    "failed logical run used a different configuration"
                );
            }
            if (!retry_failed) {
                execute(database, "COMMIT;");
                return false;
            }
        }
 
        const std::string started_at = utc_timestamp();
        auto statement = prepare(database, R"SQL(
INSERT INTO runs(
    run_id, graph_domain, logical_date, configuration_sha256, status,
    attempt_count, started_at
) VALUES (?, ?, ?, ?, 'running', 1, ?)
)SQL");
        bind_text(database, statement.get(), 1, run_id);
        bind_text(database, statement.get(), 2, graph_domain);
        bind_text(database, statement.get(), 3, logical_date);
        bind_text(database, statement.get(), 4, configuration_sha256);
        bind_text(database, statement.get(), 5, started_at);
        step_done(database, statement.get());
        auto attempt = prepare(database, R"SQL(
INSERT INTO run_attempts(run_id, attempt, status, started_at)
VALUES (?, 1, 'running', ?)
)SQL");
        bind_text(database, attempt.get(), 1, run_id);
        bind_text(database, attempt.get(), 2, started_at);
        step_done(database, attempt.get());
        execute(database, "COMMIT;");
        return true;
    } catch (...) {
        execute(database, "ROLLBACK;");
        throw;
    }
}
 
void operational_ledger::bind_product_run_inputs(
    const std::string_view run_id, const std::vector<std::string>& envelope_ids
) {
    if (!is_safe_identifier(run_id)) {
        throw std::invalid_argument("run_id must be a safe stable identifier");
    }
    if (envelope_ids.empty()) {
        throw std::invalid_argument("product run requires at least one input");
    }
    std::vector<std::string> requested = envelope_ids;
    std::ranges::sort(requested);
    if (std::ranges::adjacent_find(requested) != requested.end()) {
        throw std::invalid_argument("product run inputs must be unique");
    }
 
    sqlite3* database = impl_->database.get();
    execute(database, "BEGIN IMMEDIATE;");
    try {
        {
            auto run = prepare(database, R"SQL(
SELECT graph_domain, status FROM runs WHERE run_id=?
)SQL");
            bind_text(database, run.get(), 1, run_id);
            if (sqlite3_step(run.get()) != SQLITE_ROW) {
                throw std::logic_error("product run is unknown");
            }
            if (column_text(run.get(), 0) != "product_graph"
                || column_text(run.get(), 1) != "running") {
                throw std::logic_error(
                    "product run inputs require a running product_graph run"
                );
            }

capture_inbox_baseline()

void arachne::coordination::operational_ledger::capture_inbox_baseline

(

const std::filesystem::path &

inbox_root

)

Definition at line 903 of file coordinator.cpp.

                                                                  {
    sqlite3* database = impl_->database.get();
    auto statement = prepare(database, R"SQL(
SELECT sequence, envelope_id, from_status, to_status, actor_ref, reason,
       occurred_at
FROM state_events WHERE envelope_id = ? ORDER BY sequence
)SQL");
    bind_text(database, statement.get(), 1, envelope_id);
    std::vector<state_event> result;
    while (sqlite3_step(statement.get()) == SQLITE_ROW) {
        result.push_back(
            { sqlite3_column_int64(statement.get(), 0),
              column_text(statement.get(), 1),
              cocoon_status_from_string(column_text(statement.get(), 2)),
              cocoon_status_from_string(column_text(statement.get(), 3)),
              column_text(statement.get(), 4), column_text(statement.get(), 5),
              column_text(statement.get(), 6) }
        );
    }
    return result;
}
 
void operational_ledger::capture_inbox_baseline(
    const std::filesystem::path& inbox_root
) {
    reject_symlink_components(inbox_root, "legacy inbox path");
    const auto root = weakly_canonical_or_absolute(inbox_root);
    if (root == root.root_path()) {
        throw std::invalid_argument("legacy inbox must be a scoped directory");
    }
    sqlite3* database = impl_->database.get();
    execute(database, "BEGIN IMMEDIATE;");
    try {
        std::vector<std::filesystem::path> paths;
        for (const auto& entry :
             std::filesystem::recursive_directory_iterator(root)) {
            std::error_code status_error;
            const auto status = entry.symlink_status(status_error);
            if (status_error || std::filesystem::is_symlink(status)) {
                throw std::runtime_error(
                    "legacy inbox baseline encountered an unsafe link"
                );
            }
            if (std::filesystem::is_regular_file(status)) {
                paths.push_back(entry.path());
            }
        }
        std::ranges::sort(paths);
        for (const auto& path : paths) {
            const auto relative_path = path.lexically_relative(root);
            if (relative_path.empty() || relative_path.is_absolute()
                || !path_is_within(path, root)) {
                throw std::runtime_error(
                    "legacy inbox baseline path escaped its root"
                );
            }
            const auto relative = relative_path.generic_string();

References impl_.

Referenced by anonymous_namespace{main.cpp}::command_inbox_baseline().

Here is the caller graph for this function:

claim_logical_run()

bool arachne::coordination::operational_ledger::claim_logical_run ( std::string_view run_id, std::string_view graph_domain, std::string_view logical_date, std::string_view configuration_sha256, bool retry_failed = false, bool resume_running = false )

nodiscard

Definition at line 1115 of file coordinator.cpp.

                                                     {
        throw_sqlite(database, "read accumulation state");
    }
    accumulation_state result;
    result.accepted_count
        = static_cast<std::size_t>(sqlite3_column_int64(statement.get(), 0));
    result.accepted_bytes
        = static_cast<std::uintmax_t>(sqlite3_column_int64(statement.get(), 1));
    const auto oldest = sqlite3_column_int64(statement.get(), 2);
    if (oldest > 0) {
        result.oldest_age = std::chrono::seconds(
            std::max<std::int64_t>(0, unix_seconds() - oldest)
        );
    }
    return result;
}
 
bool operational_ledger::should_integrate(
    const accumulation_policy& policy
) const {
    const auto state = accumulation();
    return (policy.accepted_count > 0
            && state.accepted_count >= policy.accepted_count)
        || (policy.accepted_bytes > 0
            && state.accepted_bytes >= policy.accepted_bytes)
        || (policy.oldest_age.count() > 0
            && state.oldest_age >= policy.oldest_age);
}
 
bool operational_ledger::claim_logical_run(
    const std::string_view run_id, const std::string_view graph_domain,
    const std::string_view logical_date,
    const std::string_view configuration_sha256, const bool retry_failed,
    const bool resume_running
) {
    validate_run_claim(
        run_id, graph_domain, logical_date, configuration_sha256
    );
    sqlite3* database = impl_->database.get();
    execute(database, "BEGIN IMMEDIATE;");
    try {
        int prior_attempts = 0;
        std::string prior_status;
        {
            auto existing = prepare(database, R"SQL(
SELECT graph_domain, logical_date, configuration_sha256, status, attempt_count
FROM runs WHERE run_id = ?
)SQL");
            bind_text(database, existing.get(), 1, run_id);
            const int result = sqlite3_step(existing.get());
            if (result == SQLITE_ROW) {
                if (column_text(existing.get(), 0) != graph_domain
                    || column_text(existing.get(), 1) != logical_date
                    || column_text(existing.get(), 2) != configuration_sha256) {
                    throw std::logic_error(
                        "run_id is already bound to different run inputs"
                    );
                }
                prior_status = column_text(existing.get(), 3);
                prior_attempts = sqlite3_column_int(existing.get(), 4);
            } else if (result != SQLITE_DONE) {
                throw_sqlite(database, "inspect existing run");
            }
        }
        if (!prior_status.empty()) {
            if (prior_status == "running") {
                execute(database, "COMMIT;");
                return resume_running;
            }
            if (prior_status != "failed" || !retry_failed) {
                execute(database, "COMMIT;");
                return false;
            }
            {
                auto competing = prepare(database, R"SQL(
SELECT 1 FROM runs
WHERE graph_domain=? AND logical_date=? AND run_id<>?
  AND status IN ('running','succeeded') LIMIT 1
)SQL");
                bind_text(database, competing.get(), 1, graph_domain);
                bind_text(database, competing.get(), 2, logical_date);
                bind_text(database, competing.get(), 3, run_id);
                const int result = sqlite3_step(competing.get());
                if (result == SQLITE_ROW) {
                    execute(database, "COMMIT;");
                    return false;
                }
                if (result != SQLITE_DONE) {
                    throw_sqlite(database, "inspect competing logical run");
                }
            }
            const std::string started_at = utc_timestamp();
            const int next_attempt = prior_attempts + 1;
            auto retry = prepare(database, R"SQL(
UPDATE runs SET status='running', attempt_count=?, manifest_ref='',
    started_at=?, finished_at=NULL WHERE run_id=? AND status='failed'
)SQL");
            if (sqlite3_bind_int(retry.get(), 1, next_attempt) != SQLITE_OK) {
                throw_sqlite(database, "bind retry attempt");
            }
            bind_text(database, retry.get(), 2, started_at);
            bind_text(database, retry.get(), 3, run_id);
            step_done(database, retry.get());
            auto attempt = prepare(database, R"SQL(
INSERT INTO run_attempts(run_id, attempt, status, started_at)
VALUES (?, ?, 'running', ?)
)SQL");
            bind_text(database, attempt.get(), 1, run_id);
            if (sqlite3_bind_int(attempt.get(), 2, next_attempt) != SQLITE_OK) {
                throw_sqlite(database, "bind retry attempt");
            }
            bind_text(database, attempt.get(), 3, started_at);
            step_done(database, attempt.get());
            execute(database, "COMMIT;");
            return true;
        }
 
        bool date_was_claimed = false;
        std::string prior_configuration;

finish_integrated_product_run()

void arachne::coordination::operational_ledger::finish_integrated_product_run	(	std::string_view	run_id,
		std::string_view	manifest_ref )

Definition at line 1360 of file coordinator.cpp.

                                          : requested) {
            const auto envelope = get(envelope_id);
            switch (envelope.status) {
            case cocoon_status::accepted:
            case cocoon_status::waiting_processing:
            case cocoon_status::processing:
            case cocoon_status::integrated:
            case cocoon_status::failed:
                break;
            case cocoon_status::received:
            case cocoon_status::needs_format_fix:
            case cocoon_status::waiting_approval:
            case cocoon_status::rejected:
            case cocoon_status::superseded:
                throw std::logic_error(
                    "product run input is not eligible for materialization"
                );
            }
            auto insert = prepare(database, R"SQL(
INSERT INTO run_inputs(run_id, envelope_id, payload_sha256)
VALUES (?, ?, ?)
)SQL");
            bind_text(database, insert.get(), 1, run_id);
            bind_text(database, insert.get(), 2, envelope_id);
            bind_text(database, insert.get(), 3, envelope.payload_sha256);
            step_done(database, insert.get());
        }
        execute(database, "COMMIT;");
    } catch (...) {
        execute(database, "ROLLBACK;");
        throw;
    }
}
 
std::vector<envelope_record>
operational_ledger::product_run_inputs(const std::string_view run_id) const {
    if (!is_safe_identifier(run_id)) {
        throw std::invalid_argument("run_id must be a safe stable identifier");
    }
    sqlite3* database = impl_->database.get();
    {
        auto run = prepare(database, "SELECT 1 FROM runs WHERE run_id=?");
        bind_text(database, run.get(), 1, run_id);
        if (sqlite3_step(run.get()) != SQLITE_ROW) {
            throw std::out_of_range("unknown run");
        }
    }
    auto statement = prepare(database, R"SQL(
SELECT e.envelope_id, e.payload_ref, e.payload_sha256, e.format_version,
       e.submission_ref, e.title, s.status, s.accepted_by, e.supersedes,
       e.byte_length, i.payload_sha256
FROM run_inputs i
JOIN envelopes e USING(envelope_id)
JOIN envelope_state s USING(envelope_id)
WHERE i.run_id=? ORDER BY e.created_at, e.envelope_id
)SQL");
    bind_text(database, statement.get(), 1, run_id);
    std::vector<envelope_record> result;
    while (sqlite3_step(statement.get()) == SQLITE_ROW) {
        if (column_text(statement.get(), 2)
            != column_text(statement.get(), 10)) {
            throw std::logic_error("bound product input hash changed");
        }
        result.emplace_back(read_envelope(statement.get(), path_));
    }
    return result;
}
 
void operational_ledger::finish_integrated_product_run(
    const std::string_view run_id, const std::string_view manifest_ref
) {
    if (!is_safe_identifier(run_id)) {
        throw std::invalid_argument("run_id must be a safe stable identifier");
    }
    if (manifest_ref.empty()) {
        throw std::invalid_argument(
            "integrated product run requires a run manifest reference"
        );
    }
    sqlite3* database = impl_->database.get();
    execute(database, "BEGIN IMMEDIATE;");
    try {
        std::string run_status;
        std::string prior_manifest;
        int attempt_number = 0;
        {
            auto run = prepare(database, R"SQL(
SELECT graph_domain, status, attempt_count, manifest_ref
FROM runs WHERE run_id=?
)SQL");
            bind_text(database, run.get(), 1, run_id);
            if (sqlite3_step(run.get()) != SQLITE_ROW) {
                throw std::logic_error("product run is unknown");
            }
            if (column_text(run.get(), 0) != "product_graph") {
                throw std::logic_error("run is not a product_graph run");
            }
            run_status = column_text(run.get(), 1);
            attempt_number = sqlite3_column_int(run.get(), 2);
            prior_manifest = column_text(run.get(), 3);
        }
 
        std::vector<std::pair<std::string, cocoon_status>> inputs;
        {
            auto query = prepare(database, R"SQL(
SELECT i.envelope_id, s.status
FROM run_inputs i JOIN envelope_state s USING(envelope_id)
WHERE i.run_id=? ORDER BY i.envelope_id
)SQL");
            bind_text(database, query.get(), 1, run_id);
            while (sqlite3_step(query.get()) == SQLITE_ROW) {
                inputs.emplace_back(
                    column_text(query.get(), 0),
                    cocoon_status_from_string(column_text(query.get(), 1))
                );
            }
        }

finish_run()

void arachne::coordination::operational_ledger::finish_run	(	std::string_view	run_id,
		std::string_view	status,
		std::string_view	manifest_ref = {} )

Definition at line 1486 of file coordinator.cpp.

                                               : inputs) {
            (void)envelope_id;
            if (status != cocoon_status::processing
                && status != cocoon_status::integrated) {
                throw std::logic_error(
                    "product run input is neither processing nor integrated"
                );
            }
        }
        if (run_status == "succeeded") {
            if (prior_manifest != manifest_ref
                || std::ranges::any_of(inputs, [](const auto& input) {
                       return input.second != cocoon_status::integrated;
                   })) {
                throw std::logic_error(
                    "completed product run differs from reconciliation request"
                );
            }
            execute(database, "COMMIT;");
            return;
        }
        if (run_status != "running") {
            throw std::logic_error(
                "only a running product run can confirm integration"
            );
        }
 
        const std::string finished_at = utc_timestamp();
        for (const auto& [envelope_id, status] : inputs) {
            if (status == cocoon_status::integrated) {
                continue;
            }
            auto update = prepare(database, R"SQL(
UPDATE envelope_state SET status='integrated', updated_at=?
WHERE envelope_id=? AND status='processing'
)SQL");
            bind_text(database, update.get(), 1, finished_at);
            bind_text(database, update.get(), 2, envelope_id);
            step_done(database, update.get());
            if (sqlite3_changes(database) != 1) {
                throw std::logic_error(
                    "product input changed during activation reconciliation"
                );
            }
            auto event = prepare(database, R"SQL(
INSERT INTO state_events(
    envelope_id, from_status, to_status, actor_ref, reason, occurred_at
) VALUES (?, 'processing', 'integrated', 'arachne:product-reconcile', ?, ?)
)SQL");

get()

envelope_record arachne::coordination::operational_ledger::get ( std::string_view envelope_id ) const

nodiscard

Definition at line 856 of file coordinator.cpp.

history()

std::vector< state_event > arachne::coordination::operational_ledger::history ( std::string_view envelope_id ) const

nodiscard

Definition at line 885 of file coordinator.cpp.

                                                         {
    sqlite3* database = impl_->database.get();
    auto statement = prepare(
        database,
        std::string(envelope_select)
            + " WHERE s.status = ? ORDER BY e.created_at, e.envelope_id"
    );
    bind_text(database, statement.get(), 1, to_string(status));
    std::vector<envelope_record> result;
    while (sqlite3_step(statement.get()) == SQLITE_ROW) {
        result.emplace_back(read_envelope(statement.get(), path_));
    }
    return result;
}

References impl_.

intake()

envelope_record arachne::coordination::operational_ledger::intake ( const intake_request & request )

nodiscard

Definition at line 588 of file coordinator.cpp.

                                                                        {
    reject_symlink_components(request.source_path, "submission path");
    reject_symlink_components(request.inbox_root, "inbox path");
    const auto source = weakly_canonical_or_absolute(request.source_path);
    const auto inbox = weakly_canonical_or_absolute(request.inbox_root);
    if (request.submission_ref.empty() || request.title.empty()) {
        throw std::invalid_argument(
            "submission reference and title are required"
        );
    }
    if (inbox == inbox.root_path() || !inbox.has_parent_path()) {
        throw std::invalid_argument("inbox must be a scoped directory");
    }
    if (path_is_within(path_, inbox)) {
        throw std::invalid_argument(
            "operational ledger must remain outside the internal batch queue"
        );
    }
    if (legacy_inbox_root_
        && (path_is_within(inbox, *legacy_inbox_root_)
            || path_is_within(*legacy_inbox_root_, inbox))) {
        throw std::invalid_argument(
            "internal queue and external legacy inbox must not overlap"
        );
    }
    std::filesystem::create_directories(inbox);
    reject_symlink_components(inbox, "inbox path");
    const auto source_fingerprint = fingerprint_regular_file(
        source, request.max_payload_bytes, "submission"
    );
    const std::string& payload_hash = source_fingerprint.sha256;
    const std::string envelope_id = "env_"
        + crypto::sha256(
              "batch_envelope_v1\n" + payload_hash + "\n"
              + request.submission_ref
        )
              .substr(0, 32);
 
    try {
        auto existing = get(envelope_id);
        if (existing.payload_sha256 != payload_hash
            || existing.submission_ref != request.submission_ref
            || existing.title != request.title
            || existing.supersedes != request.supersedes) {
            throw std::logic_error("stable envelope identity collision");
        }
        if (!path_is_within(existing.payload_ref, inbox)) {
            throw std::runtime_error(
                "registered cocoon payload escaped the internal queue"
            );
        }
        reject_symlink_components(
            existing.payload_ref, "registered cocoon payload"
        );
        std::error_code retained_error;
        const auto retained_status = std::filesystem::symlink_status(
            existing.payload_ref, retained_error
        );
        if (retained_status.type() == std::filesystem::file_type::not_found
            || retained_error == std::errc::no_such_file_or_directory) {
            if (existing.status == cocoon_status::integrated) {
                // Successful internal-queue cleanup is intentionally durable:
                // an identical resubmission does not recreate retired bytes.
                return existing;
            }
            throw std::runtime_error(
                "queued cocoon payload disappeared before integration"
            );
        }
        const auto retained = fingerprint_regular_file(
            existing.payload_ref, request.max_payload_bytes,
            "registered cocoon payload"
        );
        if (retained.sha256 != existing.payload_sha256
            || retained.byte_length != existing.byte_length) {
            throw std::runtime_error(
                "registered internal queue payload was modified"
            );
        }
        if (existing.status == cocoon_status::received) {
            const auto after_validation = fingerprint_regular_file(
                existing.payload_ref, request.max_payload_bytes,
                "registered cocoon payload"
            );
            if (after_validation.sha256 != retained.sha256
                || after_validation.byte_length != retained.byte_length) {
                throw std::runtime_error(
                    "registered inbox payload changed during validation"
                );
            }
            constexpr auto target = cocoon_status::waiting_approval;
            try {
                return transition(
                    envelope_id, target, "arachne:intake",
                    "opaque payload received and queued for explicit approval"
                );
            } catch (const std::logic_error&) {
                existing = get(envelope_id);
                if (existing.status == target) {
                    return existing;
                }
                throw;
            }
        }
        return existing;
    } catch (const std::out_of_range&) { }
 
    std::filesystem::path payload_ref;
    file_fingerprint retained;
    if (path_is_within(source, inbox)) {
        payload_ref = source;
        retained = source_fingerprint;
    } else {
        auto normalized_request = request;
        normalized_request.inbox_root = inbox;
        payload_ref = choose_destination(normalized_request, payload_hash);
        const auto staging_root
            = inbox.parent_path() / ".arachne-intake-staging";
        reject_symlink_components(staging_root, "intake staging path");
        std::filesystem::create_directories(staging_root);
        reject_symlink_components(staging_root, "intake staging path");
        const auto sequence = intake_stage_sequence.fetch_add(1);
        const auto staging = staging_root
            / (envelope_id + "-" + std::to_string(unix_seconds()) + "-"
               + std::to_string(sequence) + ".part");
        try {
            std::filesystem::copy_file(
                source, staging, std::filesystem::copy_options::none
            );
            const auto staged = fingerprint_regular_file(
                staging, request.max_payload_bytes, "staged inbox copy"
            );
            if (staged.sha256 != payload_hash
                || staged.byte_length != source_fingerprint.byte_length) {
                throw std::runtime_error("inbox copy hash or size mismatch");
            }
            // A same-filesystem hard link publishes without an overwrite race.
            // Removing the staging name cannot modify the queued payload link.
            std::filesystem::create_hard_link(staging, payload_ref);
            if (!std::filesystem::remove(staging)) {
                throw std::runtime_error("cannot retire intake staging link");
            }
        } catch (...) {
            std::error_code ignored;
            std::filesystem::remove(staging, ignored);
            throw;
        }
        retained = fingerprint_regular_file(
            payload_ref, request.max_payload_bytes, "retained inbox payload"
        );
        if (retained.sha256 != payload_hash
            || retained.byte_length != source_fingerprint.byte_length) {
            throw std::runtime_error(
                "retained inbox payload changed during publication"
            );
        }
    }
 
    const auto final_fingerprint = fingerprint_regular_file(
        payload_ref, request.max_payload_bytes, "retained inbox payload"
    );
    if (final_fingerprint.sha256 != retained.sha256
        || final_fingerprint.byte_length != retained.byte_length) {
        throw std::runtime_error(
            "retained inbox payload changed during mechanical validation"
        );
    }
    retained = final_fingerprint;
    const auto created_at = utc_timestamp();
    sqlite3* database = impl_->database.get();
    execute(database, "BEGIN IMMEDIATE;");
    try {
        auto statement = prepare(database, R"SQL(
INSERT INTO envelopes(
    envelope_id, payload_ref, payload_sha256, format_version,
    submission_ref, title, supersedes, byte_length, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
)SQL");
        bind_text(database, statement.get(), 1, envelope_id);
        bind_text(
            database, statement.get(), 2,
            portable_payload_reference(payload_ref, path_)
        );
        bind_text(database, statement.get(), 3, payload_hash);
        if (sqlite3_bind_int(statement.get(), 4, 0) != SQLITE_OK) {
            throw_sqlite(database, "bind format version");
        }
        bind_text(database, statement.get(), 5, request.submission_ref);
        bind_text(database, statement.get(), 6, request.title);
        bind_optional(database, statement.get(), 7, request.supersedes);
        if (sqlite3_bind_int64(
                statement.get(), 8,
                static_cast<sqlite3_int64>(retained.byte_length)
            )
            != SQLITE_OK) {
            throw_sqlite(database, "bind payload length");
        }
        bind_text(database, statement.get(), 9, created_at);
        step_done(database, statement.get());
 
        auto state_statement = prepare(database, R"SQL(
INSERT INTO envelope_state(envelope_id, status, accepted_by, updated_at)
VALUES (?, 'received', NULL, ?)
)SQL");
        bind_text(database, state_statement.get(), 1, envelope_id);
        bind_text(database, state_statement.get(), 2, created_at);
        step_done(database, state_statement.get());
        execute(database, "COMMIT;");
    } catch (...) {
        execute(database, "ROLLBACK;");

References impl_, arachne::coordination::integrated, arachne::coordination::received, arachne::crypto::sha256(), arachne::coordination::intake_request::submission_ref, arachne::coordination::intake_request::title, and arachne::coordination::waiting_approval.

Referenced by anonymous_namespace{main.cpp}::command_intake().

Here is the call graph for this function:

Here is the caller graph for this function:

list()

std::vector< envelope_record > arachne::coordination::operational_ledger::list ( cocoon_status status ) const

nodiscard

Definition at line 869 of file coordinator.cpp.

                                                              {
    sqlite3* database = impl_->database.get();
    auto statement = prepare(
        database, std::string(envelope_select) + " WHERE e.envelope_id = ?"
    );
    bind_text(database, statement.get(), 1, envelope_id);
    if (sqlite3_step(statement.get()) != SQLITE_ROW) {
        throw std::out_of_range("unknown envelope");

operator=() [1/2]

operational_ledger & arachne::coordination::operational_ledger::operator= ( const operational_ledger & )

delete

operator=() [2/2]

operational_ledger & arachne::coordination::operational_ledger::operator= ( operational_ledger && )

delete

path()

const std::filesystem::path & arachne::coordination::operational_ledger::path ( ) const

nodiscardnoexcept

Definition at line 1538 of file coordinator.cpp.

product_run_inputs()

std::vector< envelope_record > arachne::coordination::operational_ledger::product_run_inputs ( std::string_view run_id ) const

nodiscard

Definition at line 1335 of file coordinator.cpp.

        {
            auto inputs = prepare(database, R"SQL(
SELECT envelope_id, payload_sha256 FROM run_inputs
WHERE run_id=? ORDER BY envelope_id
)SQL");
            bind_text(database, inputs.get(), 1, run_id);
            while (sqlite3_step(inputs.get()) == SQLITE_ROW) {
                existing.emplace_back(
                    column_text(inputs.get(), 0), column_text(inputs.get(), 1)
                );
            }
        }
 
        if (!existing.empty()) {
            if (existing.size() != requested.size()) {
                throw std::logic_error(
                    "running product run is already bound to different inputs"
                );
            }
            for (std::size_t index = 0; index < requested.size(); ++index) {
                const auto envelope = get(requested[index]);
                if (existing[index].first != requested[index]
                    || existing[index].second != envelope.payload_sha256) {

retire_queued_payload()

bool arachne::coordination::operational_ledger::retire_queued_payload ( std::string_view envelope_id, const std::filesystem::path & internal_queue_root, std::optional< std::filesystem::path > legacy_inbox_root = std::nullopt )

nodiscard

Definition at line 1007 of file coordinator.cpp.

                                            {
            result.push_back(
                { path, "recorded legacy inbox path escapes its root" }
            );
            continue;
        }
        std::error_code status_error;
        const auto status = std::filesystem::symlink_status(path, status_error);
        if (status_error || std::filesystem::is_symlink(status)
            || !std::filesystem::is_regular_file(status)) {
            result.push_back({ path, "pre-existing inbox file is missing" });
            continue;
        }
        const auto current = fingerprint_regular_file(
            path, std::numeric_limits<std::uintmax_t>::max(),
            "legacy inbox file"
        );
        if (current.byte_length != expected_size) {
            result.push_back({ path, "pre-existing inbox file size changed" });
            continue;
        }
        if (current.sha256 != expected_hash) {
            result.push_back({ path, "pre-existing inbox file hash changed" });
        }
    }
    return result;
}
 
bool operational_ledger::retire_queued_payload(
    const std::string_view envelope_id,
    const std::filesystem::path& internal_queue_root,
    std::optional<std::filesystem::path> legacy_inbox_root
) {
    reject_symlink_components(internal_queue_root, "internal queue path");
    const auto queue = weakly_canonical_or_absolute(internal_queue_root);
    if (queue == queue.root_path()) {
        throw std::invalid_argument(
            "internal queue must be a scoped directory"
        );
    }
    std::optional<std::filesystem::path> legacy = legacy_inbox_root_;
    if (legacy_inbox_root) {
        reject_symlink_components(*legacy_inbox_root, "legacy inbox path");
        const auto supplied = weakly_canonical_or_absolute(*legacy_inbox_root);
        if (legacy && supplied != *legacy) {
            throw std::invalid_argument(
                "cleanup legacy-inbox boundary differs from ledger "
                "configuration"
            );
        }
        legacy = supplied;
    }
    if (legacy
        && (path_is_within(queue, *legacy) || path_is_within(*legacy, queue))) {
        throw std::invalid_argument(
            "internal queue and external legacy inbox must be disjoint"
        );
    }
    const auto envelope = get(envelope_id);
    if (envelope.status != cocoon_status::integrated) {
        throw std::logic_error(
            "queued payload may be retired only after full integration success"
        );
    }
    reject_symlink_components(envelope.payload_ref, "queued payload path");
    if (!path_is_within(envelope.payload_ref, queue)) {
        throw std::invalid_argument(
            "refusing to retire a payload outside the internal queue"
        );
    }
    if (legacy) {
        reject_inbox_deletion_target(envelope.payload_ref, *legacy);
    }

should_integrate()

bool arachne::coordination::operational_ledger::should_integrate ( const accumulation_policy & policy ) const

nodiscard

Definition at line 1103 of file coordinator.cpp.

                                                                    {
        throw std::runtime_error(
            "cannot retire integrated queue payload: " + error.message()
        );
    }
    return true;
}
 
accumulation_state operational_ledger::accumulation() const {

transition()

envelope_record arachne::coordination::operational_ledger::transition ( std::string_view envelope_id, cocoon_status next, std::string_view actor_ref, std::string_view reason = {} )

nodiscard

Definition at line 799 of file coordinator.cpp.

                                                            :intake",
        "opaque payload received and queued for explicit approval"
    );
}
 
envelope_record operational_ledger::transition(
    const std::string_view envelope_id, const cocoon_status next,
    const std::string_view actor_ref, const std::string_view reason
) {
    if (actor_ref.empty()) {
        throw std::invalid_argument(
            "state transition requires an actor reference"
        );
    }
    sqlite3* database = impl_->database.get();
    execute(database, "BEGIN IMMEDIATE;");
    try {
        auto current_statement = prepare(database, R"SQL(
SELECT status FROM envelope_state WHERE envelope_id = ?
)SQL");
        bind_text(database, current_statement.get(), 1, envelope_id);
        if (sqlite3_step(current_statement.get()) != SQLITE_ROW) {
            throw std::out_of_range("unknown envelope");
        }
        const auto current = cocoon_status_from_string(
            column_text(current_statement.get(), 0)
        );
        if (!can_transition(current, next)) {
            throw std::logic_error(
                "invalid cocoon transition from "
                + std::string(to_string(current)) + " to "
                + std::string(to_string(next))
            );
        }
        const auto occurred_at = utc_timestamp();
        auto update = prepare(database, R"SQL(
UPDATE envelope_state
SET status = ?, accepted_by = CASE WHEN ? = 'accepted' THEN ? ELSE accepted_by END,
    updated_at = ?
WHERE envelope_id = ?
)SQL");
        bind_text(database, update.get(), 1, to_string(next));
        bind_text(database, update.get(), 2, to_string(next));
        if (next == cocoon_status::accepted) {
            bind_text(database, update.get(), 3, actor_ref);
        } else if (sqlite3_bind_null(update.get(), 3) != SQLITE_OK) {
            throw_sqlite(database, "bind accepted by");
        }
        bind_text(database, update.get(), 4, occurred_at);
        bind_text(database, update.get(), 5, envelope_id);
        step_done(database, update.get());
 

verify_inbox()

std::vector< verification_issue > arachne::coordination::operational_ledger::verify_inbox ( const std::filesystem::path & inbox_root ) const

nodiscard

Definition at line 962 of file coordinator.cpp.

                              {
                throw_sqlite(database, "bind inbox file size");
            }
            step_done(database, statement.get());
        }
        execute(database, "COMMIT;");
    } catch (...) {
        execute(database, "ROLLBACK;");
        throw;
    }
}
 
std::vector<verification_issue> operational_ledger::verify_inbox(
    const std::filesystem::path& inbox_root
) const {
    reject_symlink_components(inbox_root, "legacy inbox path");
    const auto root = weakly_canonical_or_absolute(inbox_root);
    sqlite3* database = impl_->database.get();
    auto statement = prepare(database, R"SQL(
SELECT relative_path, sha256, byte_length FROM inbox_baseline
WHERE inbox_root = ? ORDER BY relative_path
)SQL");
    bind_text(database, statement.get(), 1, root.string());
    std::vector<verification_issue> result;
    while (sqlite3_step(statement.get()) == SQLITE_ROW) {
        const std::filesystem::path relative = column_text(statement.get(), 0);
        const auto expected_hash = column_text(statement.get(), 1);
        const auto expected_size = static_cast<std::uintmax_t>(
            sqlite3_column_int64(statement.get(), 2)

Member Data Documentation

impl_

implementation* arachne::coordination::operational_ledger::impl_

private

Definition at line 141 of file coordinator.hpp.

Referenced by capture_inbox_baseline(), history(), intake(), operational_ledger(), and ~operational_ledger().

legacy_inbox_root_

std::optional<std::filesystem::path> arachne::coordination::operational_ledger::legacy_inbox_root_

private

Definition at line 143 of file coordinator.hpp.

path_

std::filesystem::path arachne::coordination::operational_ledger::path_

private

Definition at line 142 of file coordinator.hpp.

---

The documentation for this class was generated from the following files:

• coordinator.hpp
• coordinator.cpp