|
|
@@ -319,7 +319,1015 @@ bool commit_mission_state(RuntimeInstance& instance,
|
|
|
return false;
|
|
|
}
|
|
|
|
|
|
-namespace {
|
|
|
+/** Binds and restores the durable record before the opened program enters a callback. */
|
|
|
+[[nodiscard]] bool bind_mission_state(RuntimeInstance& instance, std::uint64_t now) noexcept {
|
|
|
+ mission_state::Snapshot snapshot{};
|
|
|
+ mission_state::Status status =
|
|
|
+ mission_state::bind(instance.view.binding, instance.programKey, snapshot);
|
|
|
+ ReloadAuthorization* const authorization = reload_authorization(instance.view.binding);
|
|
|
+ if (status == mission_state::Status::programMismatch && authorization != nullptr) {
|
|
|
+ status = mission_state::rebind_program(
|
|
|
+ instance.view.binding, authorization->program, instance.programKey, snapshot);
|
|
|
+ }
|
|
|
+ if (authorization != nullptr) {
|
|
|
+ *authorization = {};
|
|
|
+ }
|
|
|
+ note_vm_status(instance, "state", mission_state::status_name(status));
|
|
|
+ if (status != mission_state::Status::ready) {
|
|
|
+ log_line(
|
|
|
+ core::log::Level::warn, &instance, "state_bind", mission_state::status_name(status));
|
|
|
+ lua_vm::fault(instance.vm, "authoritative mission State binding was refused");
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ instance.missionStateBound = true;
|
|
|
+ accept_mission_state(instance, snapshot);
|
|
|
+ if (snapshot.state.variableCount > snapshot.state.variables.size()
|
|
|
+ || snapshot.state.timerCount > snapshot.state.timers.size()) {
|
|
|
+ fault_instance(instance, "authoritative mission State durable row count is invalid");
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ std::vector<lua_vm::Intent> restoredIntents{};
|
|
|
+ try {
|
|
|
+ restoredIntents.reserve(snapshot.state.pendingIntents.size());
|
|
|
+ for (const mission_state::PendingIntent& pending : snapshot.state.pendingIntents) {
|
|
|
+ restoredIntents.push_back(pending.value);
|
|
|
+ }
|
|
|
+ } catch (const std::bad_alloc&) {
|
|
|
+ fault_instance(instance, "authoritative mission State restore allocation failed");
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ if (!lua_vm::restore_state(instance.vm,
|
|
|
+ snapshot.state.phase,
|
|
|
+ snapshot.state.revision,
|
|
|
+ {snapshot.state.variables.data(), snapshot.state.variableCount},
|
|
|
+ {snapshot.state.timers.data(), snapshot.state.timerCount},
|
|
|
+ snapshot.state.nextTimerSequence,
|
|
|
+ snapshot.state.nextIntentKey,
|
|
|
+ restoredIntents)) {
|
|
|
+ fault_instance(instance, "authoritative mission State restore was refused");
|
|
|
+ log_line(core::log::Level::warn, &instance, "state_restore", "vm_refused");
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ if (instance.durableHostOutputRevision != mission_state::kAbsentHostOutputRevision) {
|
|
|
+ instance.expectedScriptableRevision = instance.durableHostOutputRevision;
|
|
|
+ instance.deliveryStage = DeliveryStage::awaitingHostCommit;
|
|
|
+ instance.deliveryDeadline = deadline_after(now, kHostCommitTimeoutMs);
|
|
|
+ instance.firstIntentAttempt = now;
|
|
|
+ instance.intentAttempts = 1;
|
|
|
+ host::InstanceSnapshot hostView{};
|
|
|
+ if (host::instance_snapshot(instance.view.binding, hostView) && hostView.outputPending
|
|
|
+ && hostView.outputKind == host::OutputKind::scriptableOverride
|
|
|
+ && hostView.scriptableRevision == instance.expectedScriptableRevision) {
|
|
|
+ instance.deliveryStage = DeliveryStage::awaitingTransport;
|
|
|
+ instance.deliveryDeadline = deadline_after(now, kTransportTimeoutMs);
|
|
|
+ } else if (hostView.scriptableRevision == instance.expectedScriptableRevision
|
|
|
+ && !hostView.outputPending
|
|
|
+ && hostView.scriptableTransportRevision != instance.expectedScriptableRevision) {
|
|
|
+ instance.deliveryDeadline = now;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (snapshot.state.faulted) {
|
|
|
+ lua_vm::fault(instance.vm, "authoritative mission State is faulted");
|
|
|
+ log_line(core::log::Level::warn, &instance, "state_restore", "faulted");
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ log_line(core::log::Level::debug,
|
|
|
+ &instance,
|
|
|
+ "state_restore",
|
|
|
+ snapshot.state.started ? "started" : "ready");
|
|
|
+ return true;
|
|
|
+}
|
|
|
+
|
|
|
+/**
|
|
|
+ * Finishes one on_load call after same-session VM reattachment.
|
|
|
+ * A restore is not a state transition, so an unchanged candidate must not spend a revision.
|
|
|
+ * @return True when the program is running.
|
|
|
+ */
|
|
|
+[[nodiscard]] bool apply_load(RuntimeInstance& instance, lua_vm::CallStatus loaded) noexcept {
|
|
|
+ lua_vm::Snapshot diagnostics{};
|
|
|
+ lua_vm::snapshot(instance.vm, diagnostics);
|
|
|
+ if (loaded != lua_vm::CallStatus::committed && loaded != lua_vm::CallStatus::noHandler) {
|
|
|
+ instance.programStatus = ProgramStatus::programError;
|
|
|
+ log_line(core::log::Level::warn,
|
|
|
+ &instance,
|
|
|
+ "load",
|
|
|
+ lua_vm::status_name(loaded),
|
|
|
+ {},
|
|
|
+ diagnostics.lastError.data());
|
|
|
+ persist_mission_fault(instance);
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ if (diagnostics.stateRevision != instance.missionStateRevision
|
|
|
+ && !commit_mission_state(instance, true, instance.lastMissionSequence)) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ instance.lastLoggedRevision = instance.missionStateRevision;
|
|
|
+ return true;
|
|
|
+}
|
|
|
+
|
|
|
+enum class InitialStateGate : std::uint8_t {
|
|
|
+ ready,
|
|
|
+ pending,
|
|
|
+ failed,
|
|
|
+};
|
|
|
+
|
|
|
+/** Selects the program-declared state once, then waits for its exact roster revision to publish. */
|
|
|
+[[nodiscard]] InitialStateGate initial_state_gate(RuntimeInstance& instance) noexcept {
|
|
|
+ if (!instance.initialStateDeclared) {
|
|
|
+ return InitialStateGate::ready;
|
|
|
+ }
|
|
|
+ activity_sdk_mission::Snapshot seed{};
|
|
|
+ const activity_sdk_mission::Status status =
|
|
|
+ instance.initialStateSelected ? activity_sdk_mission::query(instance.view, seed)
|
|
|
+ : activity_sdk_mission::select_state(
|
|
|
+ instance.view, instance.initialStateRegion, {}, seed);
|
|
|
+ if (status == activity_sdk_mission::Status::outputBusy) {
|
|
|
+ return InitialStateGate::pending;
|
|
|
+ }
|
|
|
+ if (status != activity_sdk_mission::Status::ready || !seed.configured
|
|
|
+ || seed.plan.effectiveRegion != static_cast<std::uint32_t>(instance.initialStateRegion)) {
|
|
|
+ log_line(core::log::Level::warn,
|
|
|
+ &instance,
|
|
|
+ "initial_state",
|
|
|
+ activity_sdk_mission::status_name(status));
|
|
|
+ fault_instance(instance, "program initial_state mission-seed selection was refused");
|
|
|
+ return InitialStateGate::failed;
|
|
|
+ }
|
|
|
+ instance.initialStateSelected = true;
|
|
|
+ // `plan.effectiveRegion` is the authored-state key; several authored states share one client
|
|
|
+ // slice-set region, so the client link cannot report it. The lease revision reaching the
|
|
|
+ // transport is the publication acknowledgement.
|
|
|
+ return seed.publicationPending || seed.revision == 0 || seed.publishedRevision != seed.revision
|
|
|
+ ? InitialStateGate::pending
|
|
|
+ : InitialStateGate::ready;
|
|
|
+}
|
|
|
+
|
|
|
+/** Runs and durably commits a fresh program only after its initial-state gate is open. */
|
|
|
+[[nodiscard]] bool start_program(RuntimeInstance& instance, std::uint64_t now) noexcept {
|
|
|
+ const lua_vm::CallStatus started = lua_vm::start(instance.vm, now);
|
|
|
+ note_vm_status(instance, "start", lua_vm::status_name(started));
|
|
|
+ if (started != lua_vm::CallStatus::committed && started != lua_vm::CallStatus::noHandler) {
|
|
|
+ instance.programStatus = ProgramStatus::programError;
|
|
|
+ lua_vm::Snapshot diagnostics{};
|
|
|
+ lua_vm::snapshot(instance.vm, diagnostics);
|
|
|
+ log_line(core::log::Level::warn,
|
|
|
+ &instance,
|
|
|
+ "start",
|
|
|
+ lua_vm::status_name(started),
|
|
|
+ {},
|
|
|
+ diagnostics.lastError.data());
|
|
|
+ persist_mission_fault(instance);
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ if (!commit_mission_state(instance, true, instance.lastMissionSequence)) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ instance.startPending = false;
|
|
|
+ instance.lastLoggedRevision = instance.missionStateRevision;
|
|
|
+ log_line(core::log::Level::info, &instance, "open", "ready");
|
|
|
+ return true;
|
|
|
+}
|
|
|
+
|
|
|
+/** Opens the program for one slot, binds its durable record, then reattaches or starts it. */
|
|
|
+[[nodiscard]] AttachResult open_program(RuntimeInstance& instance, std::uint64_t now) noexcept {
|
|
|
+ const format::Activity* const activity = sdk::bound_activity(instance.view);
|
|
|
+ if (activity == nullptr
|
|
|
+ || !sdk_bridge::program_identity(instance.view, instance.publicTarget, instance.identity)) {
|
|
|
+ instance.programStatus = ProgramStatus::programError;
|
|
|
+ note_vm_status(instance, "open", "invalid_sdk_view");
|
|
|
+ log_line(core::log::Level::warn, &instance, "open", "invalid_sdk_view");
|
|
|
+ return AttachResult::programError;
|
|
|
+ }
|
|
|
+ instance.identity.sdkLuaSearchPath = g_sdkLuaSearchPath;
|
|
|
+ instance.identity.playerKey = instance.playerKey;
|
|
|
+ std::span<const char> source{};
|
|
|
+ switch (read_source(*instance.view.catalog, *activity, source)) {
|
|
|
+ case SourceStatus::missing:
|
|
|
+ instance.programStatus = ProgramStatus::missing;
|
|
|
+ note_vm_status(instance, "open", "no_script");
|
|
|
+ log_line(core::log::Level::info, &instance, "open", "no_script");
|
|
|
+ return AttachResult::noScript;
|
|
|
+ case SourceStatus::fileError:
|
|
|
+ instance.programStatus = ProgramStatus::fileError;
|
|
|
+ note_vm_status(instance, "open", "file_error");
|
|
|
+ log_line(core::log::Level::warn, &instance, "open", "file_error");
|
|
|
+ return AttachResult::scriptFileError;
|
|
|
+ case SourceStatus::tooLarge:
|
|
|
+ instance.programStatus = ProgramStatus::sourceTooLarge;
|
|
|
+ note_vm_status(instance, "open", "source_too_large");
|
|
|
+ log_line(core::log::Level::warn, &instance, "open", "source_too_large");
|
|
|
+ return AttachResult::sourceTooLarge;
|
|
|
+ case SourceStatus::ready:
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ if (!make_program_key(instance, *activity, source, instance.programKey)) {
|
|
|
+ std::fill(g_source.begin(), g_source.begin() + source.size(), '\0');
|
|
|
+ instance.programStatus = ProgramStatus::programError;
|
|
|
+ note_vm_status(instance, "state", "invalid_program_key");
|
|
|
+ log_line(core::log::Level::warn, &instance, "open", "invalid_program_key");
|
|
|
+ return AttachResult::programError;
|
|
|
+ }
|
|
|
+ const lua_vm::OpenStatus opened =
|
|
|
+ lua_vm::open(instance.vm,
|
|
|
+ instance.identity,
|
|
|
+ sdk_bridge::definition_api(instance.view, instance.worldView),
|
|
|
+ source);
|
|
|
+ std::fill(g_source.begin(), g_source.begin() + source.size(), '\0');
|
|
|
+ note_vm_status(instance, "open", lua_vm::status_name(opened));
|
|
|
+ if (opened != lua_vm::OpenStatus::ready) {
|
|
|
+ instance.programStatus = ProgramStatus::programError;
|
|
|
+ lua_vm::Snapshot diagnostics{};
|
|
|
+ lua_vm::snapshot(instance.vm, diagnostics);
|
|
|
+ log_line(core::log::Level::warn,
|
|
|
+ &instance,
|
|
|
+ "open",
|
|
|
+ lua_vm::status_name(opened),
|
|
|
+ {},
|
|
|
+ diagnostics.lastError.data());
|
|
|
+ return AttachResult::programError;
|
|
|
+ }
|
|
|
+ instance.programStatus = ProgramStatus::loaded;
|
|
|
+ instance.initialStateDeclared =
|
|
|
+ lua_vm::initial_state_region(instance.vm, instance.initialStateRegion);
|
|
|
+ if (instance.initialStateDeclared) {
|
|
|
+ instance.activeRegion = instance.initialStateRegion;
|
|
|
+ }
|
|
|
+ if (!bind_mission_state(instance, now)) {
|
|
|
+ instance.programStatus = ProgramStatus::programError;
|
|
|
+ return AttachResult::programError;
|
|
|
+ }
|
|
|
+ if (instance.missionStarted) {
|
|
|
+ instance.missionReattached = true;
|
|
|
+ const lua_vm::CallStatus loaded = lua_vm::load(instance.vm, now);
|
|
|
+ note_vm_status(instance, "load", lua_vm::status_name(loaded));
|
|
|
+ if (!apply_load(instance, loaded)) {
|
|
|
+ return AttachResult::programError;
|
|
|
+ }
|
|
|
+ log_line(core::log::Level::info, &instance, "open", "ready", "reason=state_reattached");
|
|
|
+ return AttachResult::ready;
|
|
|
+ }
|
|
|
+ switch (initial_state_gate(instance)) {
|
|
|
+ case InitialStateGate::failed:
|
|
|
+ return AttachResult::programError;
|
|
|
+ case InitialStateGate::pending:
|
|
|
+ instance.startPending = true;
|
|
|
+ log_line(core::log::Level::info, &instance, "initial_state", "publication_pending");
|
|
|
+ return AttachResult::ready;
|
|
|
+ case InitialStateGate::ready:
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ return start_program(instance, now) ? AttachResult::ready : AttachResult::programError;
|
|
|
+}
|
|
|
+
|
|
|
+/** Binds one host instance to a free slot once its link, SDK view and world view all resolve. */
|
|
|
+void attach_instance(const host::InstanceSnapshot& hostInstance,
|
|
|
+ sdk::Snapshot catalog,
|
|
|
+ std::uint64_t now) noexcept {
|
|
|
+ if (find_instance(hostInstance.binding) != nullptr) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (catalog == nullptr) {
|
|
|
+ report_attach_result(
|
|
|
+ hostInstance.binding, AttachResult::catalogUnavailable, "catalog_unavailable");
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ server::bap::ActivityLinkView link{};
|
|
|
+ if (!server::bap::activity_link_view(hostInstance.binding, link)) {
|
|
|
+ report_attach_result(
|
|
|
+ hostInstance.binding, AttachResult::noActivityLink, "no_activity_link");
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (!link.joined) {
|
|
|
+ report_attach_result(
|
|
|
+ hostInstance.binding, AttachResult::noActivityLink, "activity_join_pending");
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ sdk::BoundView view{};
|
|
|
+ const sdk::Selection selection{
|
|
|
+ .binding = hostInstance.binding,
|
|
|
+ .matchingLinks = link.matchingLinks,
|
|
|
+ .activityClientGeneration = link.activityClientGeneration,
|
|
|
+ };
|
|
|
+ const sdk::Status status = sdk::resolve(catalog, selection, view);
|
|
|
+ if (status != sdk::Status::ready) {
|
|
|
+ report_attach_result(hostInstance.binding,
|
|
|
+ AttachResult::sdkStatus,
|
|
|
+ sdk::status_name(status),
|
|
|
+ format::kAbsentIndex,
|
|
|
+ status);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ generated::GeneratedWorldView worldView{};
|
|
|
+ const generated::BindStatus worldStatus = generated::resolve(view, worldView);
|
|
|
+ if (worldStatus != generated::BindStatus::ready) {
|
|
|
+ report_attach_result(hostInstance.binding,
|
|
|
+ AttachResult::generatedWorldStatus,
|
|
|
+ generated::status_name(worldStatus),
|
|
|
+ view.activityRow,
|
|
|
+ sdk::Status::notReady,
|
|
|
+ worldStatus);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ RuntimeInstance* const instance = free_instance();
|
|
|
+ if (instance == nullptr) {
|
|
|
+ report_attach_result(
|
|
|
+ hostInstance.binding, AttachResult::capacity, "capacity", view.activityRow);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ instance->view = std::move(view);
|
|
|
+ instance->worldView = std::move(worldView);
|
|
|
+ instance->publicTarget = link.publicTarget;
|
|
|
+ instance->playerKey = link.playerKey;
|
|
|
+ instance->occupied = true;
|
|
|
+ const AttachResult opened = open_program(*instance, now);
|
|
|
+ report_attach_result(
|
|
|
+ hostInstance.binding, opened, attach_result_name(opened), instance->view.activityRow);
|
|
|
+}
|
|
|
+
|
|
|
+/** Drops slots that no longer match, publishes the roster, and attaches active host instances. */
|
|
|
+void synchronize_instances(std::uint64_t now) noexcept {
|
|
|
+ host::DiagnosticsSnapshot diagnostics{};
|
|
|
+ host::snapshot(diagnostics);
|
|
|
+ retire_attach_diagnostics(diagnostics);
|
|
|
+ retire_unbound_pending_events();
|
|
|
+ for (RuntimeInstance& instance : g_instances) {
|
|
|
+ const bool bindingActive =
|
|
|
+ instance.occupied && is_active(diagnostics, instance.view.binding);
|
|
|
+ const bool bindingRetained =
|
|
|
+ instance.occupied && state::activity::binding_matches(instance.view.binding);
|
|
|
+ // A generation change only stales the view, so rebind and keep the program.
|
|
|
+ if (instance.occupied && bindingActive && bindingRetained && !still_exact(instance)
|
|
|
+ && rebind_instance(instance)) {
|
|
|
+ log_line(core::log::Level::debug, &instance, "rebind", "generation");
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (instance.occupied && (!bindingActive || !still_exact(instance))) {
|
|
|
+ log_line(core::log::Level::info, &instance, "close", "stale_generation");
|
|
|
+ // Accepted mission inputs belong to the exact SessionBinding, not one ActivityClient
|
|
|
+ // generation or one temporary link outage. Clear only after State replaces the exact
|
|
|
+ // session generation; otherwise reattach must finish every already-accepted row.
|
|
|
+ clear_instance(instance, !bindingRetained);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ std::array<state::activity::SessionRosterRow, state::activity::kSessionCapacity> roster{};
|
|
|
+ std::size_t rosterCount = 0;
|
|
|
+ static_cast<void>(state::activity::snapshot_session_roster(roster, rosterCount));
|
|
|
+ for (RuntimeInstance& instance : g_instances) {
|
|
|
+ if (instance.occupied) {
|
|
|
+ push_session_roster_edges(instance, {roster.data(), rosterCount});
|
|
|
+ }
|
|
|
+ }
|
|
|
+ const sdk::Snapshot catalog = sdk::snapshot();
|
|
|
+ for (std::size_t index = 0; index < diagnostics.instanceCount; ++index) {
|
|
|
+ if (diagnostics.instances[index].active) {
|
|
|
+ attach_instance(diagnostics.instances[index], catalog, now);
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/** Advances fresh programs only when their declared state roster has reached transport output. */
|
|
|
+void service_pending_starts(std::uint64_t now) noexcept {
|
|
|
+ for (RuntimeInstance& instance : g_instances) {
|
|
|
+ if (!instance.occupied || !instance.startPending
|
|
|
+ || instance.programStatus != ProgramStatus::loaded || instance.missionStarted) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ switch (initial_state_gate(instance)) {
|
|
|
+ case InitialStateGate::pending:
|
|
|
+ break;
|
|
|
+ case InitialStateGate::failed:
|
|
|
+ instance.startPending = false;
|
|
|
+ break;
|
|
|
+ case InitialStateGate::ready:
|
|
|
+ static_cast<void>(start_program(instance, now));
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/** @return True for the host events that report an output's progress, not an input. */
|
|
|
+[[nodiscard]] bool delivery_lifecycle_event(host::EventKind kind) noexcept {
|
|
|
+ switch (kind) {
|
|
|
+ case host::EventKind::authStateCommitted:
|
|
|
+ case host::EventKind::authStateTransportStaged:
|
|
|
+ case host::EventKind::authStateCanceled:
|
|
|
+ case host::EventKind::incidentQueued:
|
|
|
+ case host::EventKind::incidentTransportStaged:
|
|
|
+ case host::EventKind::incidentCanceled:
|
|
|
+ case host::EventKind::incidentRefused:
|
|
|
+ case host::EventKind::scriptableOverrideCommitted:
|
|
|
+ case host::EventKind::scriptableOverrideTransportStaged:
|
|
|
+ case host::EventKind::scriptableOverrideCanceled:
|
|
|
+ case host::EventKind::operatorRefused:
|
|
|
+ return true;
|
|
|
+ default:
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/**
|
|
|
+ * @return True for a row that arrives on the ordered mission-input feed and owns a sequence.
|
|
|
+ * A host-state row must never answer true. It would consume a mission-input sequence it does not
|
|
|
+ * own, which faults the binding on the next real input.
|
|
|
+ */
|
|
|
+[[nodiscard]] bool host_feed_row(host::EventKind kind) noexcept {
|
|
|
+ switch (kind) {
|
|
|
+ case host::EventKind::timerElapsed:
|
|
|
+ case host::EventKind::effectResult:
|
|
|
+ case host::EventKind::phaseEntered:
|
|
|
+ case host::EventKind::triggerEntered:
|
|
|
+ case host::EventKind::triggerExited:
|
|
|
+ case host::EventKind::squadState:
|
|
|
+ case host::EventKind::entitySpawned:
|
|
|
+ case host::EventKind::entityDied:
|
|
|
+ case host::EventKind::sceneFinished:
|
|
|
+ case host::EventKind::objectiveProgress:
|
|
|
+ case host::EventKind::sessionJoined:
|
|
|
+ case host::EventKind::sessionLeft:
|
|
|
+ case host::EventKind::playerTrigger:
|
|
|
+ case host::EventKind::cinematicStarted:
|
|
|
+ case host::EventKind::cinematicTerminated:
|
|
|
+ return false;
|
|
|
+ default:
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/** True when the event may reach a callback for this instance's ActivityClient generation. */
|
|
|
+[[nodiscard]] bool eligible_event(const RuntimeInstance& instance,
|
|
|
+ const host::Event& event) noexcept {
|
|
|
+ if (event.kind == host::EventKind::timerElapsed) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ if (event.sourceGeneration != instance.view.activityClientGeneration) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ return event.kind == host::EventKind::clientStateChanged
|
|
|
+ || event.kind == host::EventKind::incidentReceived
|
|
|
+ || event.kind == host::EventKind::clientMessageReceived
|
|
|
+ || event.kind == host::EventKind::effectResult
|
|
|
+ || event.kind == host::EventKind::phaseEntered
|
|
|
+ || event.kind == host::EventKind::triggerEntered
|
|
|
+ || event.kind == host::EventKind::triggerExited
|
|
|
+ || event.kind == host::EventKind::squadState
|
|
|
+ || event.kind == host::EventKind::entitySpawned
|
|
|
+ || event.kind == host::EventKind::entityDied
|
|
|
+ || event.kind == host::EventKind::sceneFinished
|
|
|
+ || event.kind == host::EventKind::objectiveProgress
|
|
|
+ || event.kind == host::EventKind::entitySlotsRequested
|
|
|
+ || event.kind == host::EventKind::sessionJoined
|
|
|
+ || event.kind == host::EventKind::sessionLeft
|
|
|
+ || event.kind == host::EventKind::playerTrigger
|
|
|
+ || event.kind == host::EventKind::cinematicStarted
|
|
|
+ || event.kind == host::EventKind::cinematicTerminated
|
|
|
+ || delivery_lifecycle_event(event.kind)
|
|
|
+ || (event.kind == host::EventKind::senseUpdate
|
|
|
+ && event.senseDecodeStatus
|
|
|
+ == middleware::bap::activity_message::sense_update::DecodeStatus::complete);
|
|
|
+}
|
|
|
+
|
|
|
+/** Faults the instance unless the ordered mission input arrives with no gap, starting at one. */
|
|
|
+[[nodiscard]] bool validate_mission_sequence(RuntimeInstance& instance,
|
|
|
+ const host::Event& event) noexcept {
|
|
|
+ if (event.kind != host::EventKind::senseUpdate
|
|
|
+ && event.kind != host::EventKind::incidentReceived
|
|
|
+ && event.kind != host::EventKind::clientStateChanged
|
|
|
+ && event.kind != host::EventKind::entitySlotsRequested
|
|
|
+ && event.kind != host::EventKind::clientMessageReceived) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ if (instance.lastMissionSequence == 0) {
|
|
|
+ if (event.missionSequence == 1) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ fault_instance(instance, "activity mission input did not start at sequence one");
|
|
|
+ log_line(core::log::Level::warn, &instance, "events", "initial_binding_gap");
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ const std::uint64_t expected =
|
|
|
+ instance.lastMissionSequence == (std::numeric_limits<std::uint64_t>::max)()
|
|
|
+ ? 1
|
|
|
+ : instance.lastMissionSequence + 1;
|
|
|
+ if (event.missionSequence == expected) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ fault_instance(instance, "activity mission input sequence has a gap");
|
|
|
+ log_line(core::log::Level::warn, &instance, "events", "binding_gap");
|
|
|
+ return false;
|
|
|
+}
|
|
|
+
|
|
|
+/** Records the typed Ghost-link fields while its interaction semantics are being verified. */
|
|
|
+void log_ghost_link_sense(const RuntimeInstance& instance,
|
|
|
+ const host::SenseObservationSnapshot& sense) noexcept {
|
|
|
+ for (std::size_t index = 0; index < sense.observationCount; ++index) {
|
|
|
+ const auto& observation = sense.observations[index];
|
|
|
+ if (observation.key.slotType != 65 || observation.key.senseSchema != 0x80804D3EU
|
|
|
+ || observation.firstValue > sense.valueCount
|
|
|
+ || observation.valueCount > sense.valueCount - observation.firstValue) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ for (std::size_t offset = 0; offset < observation.valueCount; ++offset) {
|
|
|
+ const auto& value = sense.values[observation.firstValue + offset];
|
|
|
+ if (!value.present || value.schemaRow != observation.key.schemaRow) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ std::array<char, 192> fields{};
|
|
|
+ const int written = std::snprintf(
|
|
|
+ fields.data(),
|
|
|
+ fields.size(),
|
|
|
+ "registry=%08x object=%08x slot=%u ordinal=%u bits=%u value=%llu sequence=%llu",
|
|
|
+ observation.key.registryKey,
|
|
|
+ observation.key.objectTag,
|
|
|
+ static_cast<unsigned>(observation.key.slotIndex),
|
|
|
+ static_cast<unsigned>(value.fieldOrdinal),
|
|
|
+ static_cast<unsigned>(value.width),
|
|
|
+ static_cast<unsigned long long>(value.unsignedValue),
|
|
|
+ static_cast<unsigned long long>(observation.sequence));
|
|
|
+ if (written > 0) {
|
|
|
+ log_line(core::log::Level::info,
|
|
|
+ &instance,
|
|
|
+ "ghost_link_sense",
|
|
|
+ "observed",
|
|
|
+ {fields.data(),
|
|
|
+ (std::min)(static_cast<std::size_t>(written), fields.size() - 1)});
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/** Runs one event through the VM, commits what it changed, and faults on a script failure. */
|
|
|
+[[nodiscard]] lua_vm::CallStatus dispatch_event(RuntimeInstance& instance,
|
|
|
+ const host::Event& event,
|
|
|
+ const host::SenseObservationSnapshot* sense,
|
|
|
+ const host::ClientMessageSnapshot* clientMessage,
|
|
|
+ bool firstAttempt,
|
|
|
+ std::uint64_t now) noexcept {
|
|
|
+ if (instance.programStatus == ProgramStatus::missing
|
|
|
+ && event.kind == host::EventKind::senseUpdate && sense != nullptr) {
|
|
|
+ push_squad_edges(instance, *sense);
|
|
|
+ return lua_vm::CallStatus::inactive;
|
|
|
+ }
|
|
|
+ if (instance.programStatus != ProgramStatus::loaded || !eligible_event(instance, event)) {
|
|
|
+ return lua_vm::CallStatus::inactive;
|
|
|
+ }
|
|
|
+ if (firstAttempt) {
|
|
|
+ ++instance.eventsSeen;
|
|
|
+ instance.lastEventSequence = event.sequence;
|
|
|
+ }
|
|
|
+ // The three region numbers the script is about to read. A report restates only the leg it
|
|
|
+ // moved, so `pending` and `current` read -1 on most reports and `held` is the one that says
|
|
|
+ // where the client is standing.
|
|
|
+ if (firstAttempt && event.kind == host::EventKind::clientStateChanged) {
|
|
|
+ std::array<char, 64> legs{};
|
|
|
+ const int written =
|
|
|
+ std::snprintf(legs.data(),
|
|
|
+ legs.size(),
|
|
|
+ "held=%d pending=%d current=%d",
|
|
|
+ event.heldRegionIndex,
|
|
|
+ event.clientStateHasRegion ? event.regionIndex : -1,
|
|
|
+ event.clientStateHasCurrentRegion ? event.currentRegionIndex : -1);
|
|
|
+ if (written > 0) {
|
|
|
+ log_line(core::log::Level::debug,
|
|
|
+ &instance,
|
|
|
+ "client_state",
|
|
|
+ "legs",
|
|
|
+ {legs.data(), static_cast<std::size_t>(written)});
|
|
|
+ }
|
|
|
+ }
|
|
|
+ const lua_vm::CallStatus status = lua_vm::dispatch(instance.vm, event, clientMessage, now);
|
|
|
+ if (event.kind == host::EventKind::clientStateChanged && event.clientStateHasRegion) {
|
|
|
+ instance.activeRegion = event.regionIndex;
|
|
|
+ }
|
|
|
+ if (firstAttempt && event.kind == host::EventKind::incidentReceived) {
|
|
|
+ push_player_trigger(instance, event);
|
|
|
+ push_cinematic(instance, event);
|
|
|
+ }
|
|
|
+ if (event.kind == host::EventKind::senseUpdate && sense != nullptr) {
|
|
|
+ if (firstAttempt) {
|
|
|
+ log_ghost_link_sense(instance, *sense);
|
|
|
+ }
|
|
|
+ push_trigger_edges(instance, *sense);
|
|
|
+ push_squad_edges(instance, *sense);
|
|
|
+ push_scene_edges(instance, *sense);
|
|
|
+ push_objective_edges(instance, *sense);
|
|
|
+ }
|
|
|
+ note_vm_status(instance, "event", lua_vm::status_name(status));
|
|
|
+ if (firstAttempt && instance.eventsSeen == 1) {
|
|
|
+ log_line(core::log::Level::info,
|
|
|
+ &instance,
|
|
|
+ "dispatch",
|
|
|
+ lua_vm::status_name(status),
|
|
|
+ event.kind == host::EventKind::senseUpdate ? "sense"
|
|
|
+ : event.kind == host::EventKind::clientStateChanged ? "client_state"
|
|
|
+ : event.kind == host::EventKind::incidentReceived ? "incident"
|
|
|
+ : event.kind == host::EventKind::entitySlotsRequested ? "entity_slots_requested"
|
|
|
+ : event.kind == host::EventKind::timerElapsed ? "timer"
|
|
|
+ : event.kind == host::EventKind::effectResult ? "effect_result"
|
|
|
+ : "client_message");
|
|
|
+ }
|
|
|
+ const std::uint64_t nextInputSequence =
|
|
|
+ host_feed_row(event.kind) ? event.missionSequence : instance.lastMissionSequence;
|
|
|
+ if (status == lua_vm::CallStatus::committed) {
|
|
|
+ if (!commit_mission_state(instance, true, nextInputSequence)) {
|
|
|
+ return lua_vm::CallStatus::scriptError;
|
|
|
+ }
|
|
|
+ ++instance.eventsCommitted;
|
|
|
+ lua_vm::Snapshot diagnostics{};
|
|
|
+ lua_vm::snapshot(instance.vm, diagnostics);
|
|
|
+ if (event.kind == host::EventKind::clientStateChanged
|
|
|
+ || diagnostics.stateRevision != instance.lastLoggedRevision) {
|
|
|
+ instance.lastLoggedRevision = diagnostics.stateRevision;
|
|
|
+ log_line(core::log::Level::debug,
|
|
|
+ &instance,
|
|
|
+ "event",
|
|
|
+ "committed",
|
|
|
+ event.kind == host::EventKind::senseUpdate ? "sense"
|
|
|
+ : event.kind == host::EventKind::clientStateChanged ? "client_state"
|
|
|
+ : event.kind == host::EventKind::incidentReceived ? "incident"
|
|
|
+ : event.kind == host::EventKind::entitySlotsRequested
|
|
|
+ ? "entity_slots_requested"
|
|
|
+ : event.kind == host::EventKind::timerElapsed ? "timer"
|
|
|
+ : event.kind == host::EventKind::effectResult ? "effect_result"
|
|
|
+ : "client_message");
|
|
|
+ }
|
|
|
+ return status;
|
|
|
+ }
|
|
|
+ if (status == lua_vm::CallStatus::noHandler) {
|
|
|
+ return commit_mission_state(instance, true, nextInputSequence)
|
|
|
+ ? status
|
|
|
+ : lua_vm::CallStatus::scriptError;
|
|
|
+ }
|
|
|
+ if (status == lua_vm::CallStatus::inactive) {
|
|
|
+ return status;
|
|
|
+ }
|
|
|
+ lua_vm::Snapshot diagnostics{};
|
|
|
+ lua_vm::snapshot(instance.vm, diagnostics);
|
|
|
+ log_line(core::log::Level::warn,
|
|
|
+ &instance,
|
|
|
+ "event",
|
|
|
+ lua_vm::status_name(status),
|
|
|
+ {},
|
|
|
+ diagnostics.lastError.data());
|
|
|
+ persist_mission_fault(instance);
|
|
|
+ return status;
|
|
|
+}
|
|
|
+
|
|
|
+/** Reads new host events, advances delivery, and queues only the lifecycle rows for scripts. */
|
|
|
+void consume_delivery_events(std::uint64_t now) noexcept {
|
|
|
+ host::EventRead events{};
|
|
|
+ host::read_events_after(g_eventCursor, events);
|
|
|
+ g_eventCursor = events.cursor;
|
|
|
+ if (events.reset) {
|
|
|
+ log_line(core::log::Level::warn, nullptr, "delivery", "event_feed_reset");
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (events.gap) {
|
|
|
+ log_line(core::log::Level::warn, nullptr, "delivery", "event_feed_gap");
|
|
|
+ }
|
|
|
+ for (std::size_t index = 0; index < events.count; ++index) {
|
|
|
+ RuntimeInstance* const instance = find_instance(events.events[index].binding);
|
|
|
+ if (instance != nullptr) {
|
|
|
+ observe_delivery_event(*instance, events.events[index], now);
|
|
|
+ // Sense, client-state, incident and client-message rows reach the script through the
|
|
|
+ // ordered mission-input feed. Only the delivery lifecycle rows belong in this queue.
|
|
|
+ if (delivery_lifecycle_event(events.events[index].kind)) {
|
|
|
+ push_script_event(*instance, events.events[index]);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/** One free queue row, growing the queue by one when none is free; null when it cannot grow. */
|
|
|
+[[nodiscard]] PendingMissionEvent* free_pending_event() noexcept {
|
|
|
+ for (PendingMissionEvent& pending : g_pendingMissionEvents) {
|
|
|
+ if (!pending.occupied) {
|
|
|
+ return &pending;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (g_pendingMissionEvents.size() == g_pendingMissionEvents.max_size()) {
|
|
|
+ return nullptr;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ g_pendingMissionEvents.emplace_back();
|
|
|
+ } catch (const std::bad_alloc&) {
|
|
|
+ return nullptr;
|
|
|
+ }
|
|
|
+ return &g_pendingMissionEvents.back();
|
|
|
+}
|
|
|
+
|
|
|
+/** Copies one accepted input and its values into a queue row; faults the instance when full. */
|
|
|
+[[nodiscard]] bool queue_mission_event(RuntimeInstance* instance,
|
|
|
+ const host::MissionInputEvent& input) noexcept {
|
|
|
+ PendingMissionEvent* const pending = free_pending_event();
|
|
|
+ if (pending == nullptr) {
|
|
|
+ if (instance != nullptr) {
|
|
|
+ fault_instance(*instance, "mission event queue allocation failed");
|
|
|
+ clear_pending_events(instance->view.binding);
|
|
|
+ }
|
|
|
+ log_line(core::log::Level::warn, instance, "events", "allocation_failed");
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ clear_pending_event(*pending);
|
|
|
+ if (input.event.kind == host::EventKind::senseUpdate
|
|
|
+ && input.event.senseDecodeStatus
|
|
|
+ == middleware::bap::activity_message::sense_update::DecodeStatus::complete) {
|
|
|
+ pending->senseAvailable =
|
|
|
+ host::mission_input_sense_snapshot(input.sequence, pending->sense);
|
|
|
+ }
|
|
|
+ if (input.event.kind == host::EventKind::clientMessageReceived) {
|
|
|
+ pending->clientMessageAvailable =
|
|
|
+ host::mission_input_client_message_snapshot(input.sequence, pending->clientMessage);
|
|
|
+ }
|
|
|
+ pending->nextAttempt = 0;
|
|
|
+ pending->event = input.event;
|
|
|
+ if (instance != nullptr) {
|
|
|
+ pending->callbackEligible = eligible_event(*instance, input.event);
|
|
|
+ pending->eligibilityResolved = true;
|
|
|
+ }
|
|
|
+ pending->occupied = true;
|
|
|
+ return true;
|
|
|
+}
|
|
|
+
|
|
|
+[[nodiscard]] constexpr bool mission_sequence_precedes(std::uint64_t left,
|
|
|
+ std::uint64_t right) noexcept {
|
|
|
+ constexpr std::uint64_t halfRange = std::uint64_t{1} << 63U;
|
|
|
+ return left != right && right - left < halfRange;
|
|
|
+}
|
|
|
+
|
|
|
+static_assert(mission_sequence_precedes((std::numeric_limits<std::uint64_t>::max)(), 1));
|
|
|
+static_assert(!mission_sequence_precedes(1, (std::numeric_limits<std::uint64_t>::max)()));
|
|
|
+
|
|
|
+/** @return True when the durable cursor already committed this retained input row. */
|
|
|
+[[nodiscard]] bool mission_sequence_committed(std::uint64_t sequence,
|
|
|
+ std::uint64_t committed) noexcept {
|
|
|
+ return committed != 0
|
|
|
+ && (sequence == committed || mission_sequence_precedes(sequence, committed));
|
|
|
+}
|
|
|
+
|
|
|
+/** True when the same binding still holds a queued row with an earlier mission sequence. */
|
|
|
+[[nodiscard]] bool has_earlier_pending_event(const PendingMissionEvent& selected) noexcept {
|
|
|
+ for (const PendingMissionEvent& pending : g_pendingMissionEvents) {
|
|
|
+ if (pending.occupied && &pending != &selected
|
|
|
+ && same_binding(pending.event.binding, selected.event.binding)
|
|
|
+ && mission_sequence_precedes(pending.event.missionSequence,
|
|
|
+ selected.event.missionSequence)) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return false;
|
|
|
+}
|
|
|
+
|
|
|
+/** TODO: no caller. Decide whether the drain gate retires on this or on `binding_matches`. */
|
|
|
+[[nodiscard]] bool host_binding_active(const state::activity::SessionBinding& binding) noexcept {
|
|
|
+ host::InstanceSnapshot snapshot{};
|
|
|
+ return host::instance_snapshot(binding, snapshot) && snapshot.active;
|
|
|
+}
|
|
|
+
|
|
|
+/** Dispatches queued rows in mission-sequence order and retires those no callback can take. */
|
|
|
+void drain_pending_mission_events(std::uint64_t now) noexcept {
|
|
|
+ bool progressed = false;
|
|
|
+ do {
|
|
|
+ progressed = false;
|
|
|
+ for (PendingMissionEvent& pending : g_pendingMissionEvents) {
|
|
|
+ if (!pending.occupied || now < pending.nextAttempt
|
|
|
+ || has_earlier_pending_event(pending)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ RuntimeInstance* const instance = find_instance(pending.event.binding);
|
|
|
+ if (instance == nullptr) {
|
|
|
+ if (!state::activity::binding_matches(pending.event.binding)) {
|
|
|
+ clear_pending_event(pending);
|
|
|
+ progressed = true;
|
|
|
+ }
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (instance->programStatus != ProgramStatus::loaded) {
|
|
|
+ clear_pending_event(pending);
|
|
|
+ progressed = true;
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (instance->startPending) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (instance->timerPending) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (mission_sequence_committed(pending.event.missionSequence,
|
|
|
+ instance->lastMissionSequence)) {
|
|
|
+ clear_pending_event(pending);
|
|
|
+ progressed = true;
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (!pending.missionSequenceObserved) {
|
|
|
+ if (!validate_mission_sequence(*instance, pending.event)) {
|
|
|
+ clear_pending_events(instance->view.binding);
|
|
|
+ progressed = true;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ pending.missionSequenceObserved = true;
|
|
|
+ }
|
|
|
+ if (!pending.eligibilityResolved) {
|
|
|
+ pending.callbackEligible = eligible_event(*instance, pending.event);
|
|
|
+ pending.eligibilityResolved = true;
|
|
|
+ }
|
|
|
+ if (!pending.callbackEligible) {
|
|
|
+ if (!commit_mission_state(
|
|
|
+ *instance, instance->missionStarted, pending.event.missionSequence)) {
|
|
|
+ clear_pending_events(instance->view.binding);
|
|
|
+ progressed = true;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ clear_pending_event(pending);
|
|
|
+ progressed = true;
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (pending.event.kind == host::EventKind::senseUpdate && !pending.senseAvailable) {
|
|
|
+ fault_instance(*instance, "accepted mission Sense values were unavailable");
|
|
|
+ clear_pending_events(instance->view.binding);
|
|
|
+ log_line(core::log::Level::warn, instance, "events", "sense_unavailable");
|
|
|
+ progressed = true;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ if (pending.event.kind == host::EventKind::clientMessageReceived
|
|
|
+ && !pending.clientMessageAvailable) {
|
|
|
+ fault_instance(*instance,
|
|
|
+ "accepted mission client-message values were unavailable");
|
|
|
+ clear_pending_events(instance->view.binding);
|
|
|
+ log_line(core::log::Level::warn, instance, "events", "client_message_unavailable");
|
|
|
+ progressed = true;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ lua_vm::Intent intent{};
|
|
|
+ if (instance->deliveryStage != DeliveryStage::idle
|
|
|
+ || lua_vm::pending_intent(instance->vm, intent)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ const bool firstAttempt = pending.attempts == 0;
|
|
|
+ if (firstAttempt) {
|
|
|
+ pending.firstAttempt = now;
|
|
|
+ }
|
|
|
+ ++pending.attempts;
|
|
|
+ const host::SenseObservationSnapshot* const sense =
|
|
|
+ pending.event.kind == host::EventKind::senseUpdate && pending.senseAvailable
|
|
|
+ ? &pending.sense
|
|
|
+ : nullptr;
|
|
|
+ const host::ClientMessageSnapshot* const clientMessage =
|
|
|
+ pending.event.kind == host::EventKind::clientMessageReceived
|
|
|
+ && pending.clientMessageAvailable
|
|
|
+ ? &pending.clientMessage
|
|
|
+ : nullptr;
|
|
|
+ static_cast<void>(
|
|
|
+ dispatch_event(*instance, pending.event, sense, clientMessage, firstAttempt, now));
|
|
|
+ clear_pending_event(pending);
|
|
|
+ progressed = true;
|
|
|
+ }
|
|
|
+ } while (progressed);
|
|
|
+}
|
|
|
+
|
|
|
+/** @return True when one exact accepted sequence is already retained locally or in this read. */
|
|
|
+[[nodiscard]] bool input_sequence_retained(const state::activity::SessionBinding& binding,
|
|
|
+ std::uint64_t sequence,
|
|
|
+ const host::MissionInputRead& inputs) noexcept {
|
|
|
+ for (const PendingMissionEvent& pending : g_pendingMissionEvents) {
|
|
|
+ if (pending.occupied && same_binding(pending.event.binding, binding)
|
|
|
+ && pending.event.missionSequence == sequence) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ for (std::size_t index = 0; index < inputs.count; ++index) {
|
|
|
+ if (same_binding(inputs.events[index].event.binding, binding)
|
|
|
+ && inputs.events[index].event.missionSequence == sequence) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return false;
|
|
|
+}
|
|
|
+
|
|
|
+/** @return True when every accepted but uncommitted sequence is present in this read or local
|
|
|
+ * queue. */
|
|
|
+[[nodiscard]] bool
|
|
|
+outstanding_input_interval_complete(const state::activity::SessionBinding& binding,
|
|
|
+ const mission_state::InputSequenceSnapshot& state,
|
|
|
+ const host::MissionInputRead& inputs) noexcept {
|
|
|
+ if (state.issued < state.committed) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ const std::uint64_t outstanding = state.issued - state.committed;
|
|
|
+ if (outstanding > g_pendingMissionEvents.size() + inputs.count) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ std::uint64_t sequence = state.committed;
|
|
|
+ for (std::uint64_t index = 0; index < outstanding; ++index) {
|
|
|
+ ++sequence;
|
|
|
+ if (!input_sequence_retained(binding, sequence, inputs)) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return true;
|
|
|
+}
|
|
|
+
|
|
|
+/** Faults only retained bindings whose durable uncommitted interval is provably incomplete. */
|
|
|
+void reconcile_input_feed_loss(const host::MissionInputRead& inputs) noexcept {
|
|
|
+ host::DiagnosticsSnapshot hostState{};
|
|
|
+ host::snapshot(hostState);
|
|
|
+ for (std::size_t index = 0; index < hostState.instanceCount; ++index) {
|
|
|
+ const host::InstanceSnapshot& hostInstance = hostState.instances[index];
|
|
|
+ if (!state::activity::binding_matches(hostInstance.binding)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ mission_state::InputSequenceSnapshot inputState{};
|
|
|
+ if (!mission_state::input_sequence_snapshot(hostInstance.binding, inputState)
|
|
|
+ || inputState.faulted
|
|
|
+ || outstanding_input_interval_complete(hostInstance.binding, inputState, inputs)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ mission_state::Snapshot snapshot{};
|
|
|
+ const mission_state::Status status =
|
|
|
+ mission_state::fault_input_feed(hostInstance.binding, snapshot);
|
|
|
+ RuntimeInstance* const instance = find_instance(hostInstance.binding);
|
|
|
+ if (status != mission_state::Status::ready) {
|
|
|
+ log_line(core::log::Level::warn,
|
|
|
+ instance,
|
|
|
+ "events",
|
|
|
+ mission_state::status_name(status),
|
|
|
+ "reason=feed_gap_fault_refused");
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (instance != nullptr) {
|
|
|
+ accept_mission_state(*instance, snapshot);
|
|
|
+ lua_vm::fault(instance->vm, "accepted mission input feed lost a row");
|
|
|
+ instance->programStatus = ProgramStatus::programError;
|
|
|
+ }
|
|
|
+ log_line(core::log::Level::warn, instance, "events", "feed_gap_faulted");
|
|
|
+ clear_pending_events(hostInstance.binding);
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/** Reads one page of the ordered feed and queues every row not yet committed. */
|
|
|
+[[nodiscard]] bool consume_mission_input_page() noexcept {
|
|
|
+ host::MissionInputRead inputs{};
|
|
|
+ host::read_mission_inputs_after(g_missionInputCursor, inputs);
|
|
|
+ if (inputs.reset) {
|
|
|
+ g_missionInputCursor = {inputs.cursor.generation, 0};
|
|
|
+ reconcile_input_feed_loss(inputs);
|
|
|
+ log_line(core::log::Level::warn, nullptr, "events", "mission_feed_reset");
|
|
|
+ }
|
|
|
+ if (inputs.gap) {
|
|
|
+ log_line(core::log::Level::warn, nullptr, "events", "mission_feed_gap");
|
|
|
+ reconcile_input_feed_loss(inputs);
|
|
|
+ }
|
|
|
+ for (std::size_t index = 0; index < inputs.count; ++index) {
|
|
|
+ const host::MissionInputEvent& input = inputs.events[index];
|
|
|
+ RuntimeInstance* const instance = find_instance(input.event.binding);
|
|
|
+ mission_state::InputSequenceSnapshot inputState{};
|
|
|
+ const bool hasInputState =
|
|
|
+ mission_state::input_sequence_snapshot(input.event.binding, inputState);
|
|
|
+ if ((instance == nullptr && !state::activity::binding_matches(input.event.binding))
|
|
|
+ || (instance != nullptr && instance->programStatus != ProgramStatus::loaded)
|
|
|
+ || (hasInputState
|
|
|
+ && (inputState.faulted
|
|
|
+ || mission_sequence_committed(input.event.missionSequence,
|
|
|
+ inputState.committed)))) {
|
|
|
+ g_missionInputCursor.generation = inputs.cursor.generation;
|
|
|
+ g_missionInputCursor.sequence = input.sequence;
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (instance != nullptr
|
|
|
+ && mission_sequence_committed(input.event.missionSequence,
|
|
|
+ instance->lastMissionSequence)) {
|
|
|
+ g_missionInputCursor.generation = inputs.cursor.generation;
|
|
|
+ g_missionInputCursor.sequence = input.sequence;
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (instance != nullptr) {
|
|
|
+ lua_vm::Snapshot diagnostics{};
|
|
|
+ lua_vm::snapshot(instance->vm, diagnostics);
|
|
|
+ if (diagnostics.faulted) {
|
|
|
+ g_missionInputCursor.generation = inputs.cursor.generation;
|
|
|
+ g_missionInputCursor.sequence = input.sequence;
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (!queue_mission_event(instance, input)) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ g_missionInputCursor.generation = inputs.cursor.generation;
|
|
|
+ g_missionInputCursor.sequence = input.sequence;
|
|
|
+ }
|
|
|
+ return inputs.count == host::kMissionInputReadPageSize;
|
|
|
+}
|
|
|
+
|
|
|
+/**
|
|
|
+ * Reads the ordered mission-input feed and drains the queue. The feed is unbounded and one read
|
|
|
+ * copies at most a page, so a burst is consumed in the tick it arrives instead of a page a tick.
|
|
|
+ */
|
|
|
+void consume_mission_inputs(std::uint64_t now) noexcept {
|
|
|
+ drain_pending_mission_events(now);
|
|
|
+ while (consume_mission_input_page()) {
|
|
|
+ drain_pending_mission_events(now);
|
|
|
+ }
|
|
|
+ drain_pending_mission_events(now);
|
|
|
+}
|
|
|
|
|
|
/**
|
|
|
* Settles the inputs of every instance whose activity has no script. No program will consume
|
Millie