fix: recover instead of throw on dup schedules

Author: michael-0acf4

src/substantial/src/converters.rs

@@ -1,7 +1,7 @@
 // Copyright Metatype OÜ, licensed under the Mozilla Public License Version 2.0.
 // SPDX-License-Identifier: MPL-2.0
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use anyhow::{bail, Context, Ok, Result};
 use chrono::{DateTime, Utc};
@@ -104,12 +104,15 @@ impl Run {
                     format!("Recovering operations from backend for {}", self.run_id)
                 })?;
             self.operations = operations;
+            self.compact();
         }
 
         Ok(())
     }
 
-    pub fn persist_into(&self, backend: &dyn Backend) -> Result<()> {
+    pub fn persist_into(&mut self, backend: &dyn Backend) -> Result<()> {
+        self.compact();
+
         let mut records = Records::new();
         records.events = self
             .operations
@@ -128,6 +131,35 @@ impl Run {
     pub fn reset(&mut self) {
         self.operations = vec![];
     }
+
+    /// Try to recover from dups such as
+    ///
+    /// ```
+    ///  [Start Start ...] or
+    ///  [... Stop Stop] or
+    ///  [ .... Event X Event X ... ]
+    /// ```
+    ///
+    /// These dups can occur when we crash at a given timing
+    /// and the underlying event of the appointed schedule was not closed.
+    /// The engine will happily append onto the operation log,
+    /// we throw by default but realistically we can recover.
+    ///
+    /// WARN: undesirable side effects can still happen if we crash before saving the Saved results.
+    pub fn compact(&mut self) {
+        let mut operations = Vec::new();
+        let mut dup_schedules = HashSet::new();
+        for operation in &self.operations {
+            if dup_schedules.contains(&operation.at) {
+                continue;
+            }
+
+            operations.push(operation.clone());
+            dup_schedules.insert(operation.at);
+        }
+
+        self.operations = operations;
+    }
 }
 
 impl TryFrom<Event> for Operation {

src/typegate/engine/src/runtimes/substantial.rs

@@ -77,7 +77,7 @@ pub async fn op_sub_store_create_or_get_run(
     Ok(CreateOrGetOutput { run })
 }
 
-#[derive(Deserialize, Debug)]
+#[derive(Deserialize, Debug, Clone)]
 pub struct PersistRunInput {
     pub run: Run,
     pub backend: SubstantialBackend,
@@ -95,6 +95,8 @@ pub async fn op_sub_store_persist_run(
         state.borrow::<Ctx>().clone()
     };
 
+    let mut input = input.clone();
+
     let backend = ctx
         .backends
         .entry(input.backend.as_key())

src/typegate/src/runtimes/substantial/agent.ts

@@ -227,9 +227,22 @@ export class Agent {
     // worker matching the runId.
     if (this.workerManager.isOngoing(next.run_id)) {
       this.logger.warn(
-        `skip triggering ${next.run_id} for the current tick as it is still ongoing`,
+        `skip triggering ${next.run_id} for the current tick as it is still ongoing, will renew the lease`,
       );
 
+      // REMOVE_ME: once an a better lock approach is devised
+      // Ideally, we want to renew the lease but this might hide other timing bugs
+      // Hypothesis:
+      //   lease timing is such that when 2+ agents fights for it, one holds onto it and runs,
+      //   but one amoung the others, there is one that is still in process of freeing up its ressource since deallocation != un-leased
+      // FIX: hearbeat lease, but even with that we need a strong guarantee that ressource-freeing up == un-leased
+      //      timing bugs can still occur but more rare
+      // await Meta.substantial.agentRenewLease({
+      //   backend: this.backend,
+      //   lease_seconds: this.config.leaseLifespanSec,
+      //   run_id: next.run_id,
+      // });
+
       return;
     }
 
@@ -419,7 +432,7 @@ export class Agent {
       schedule: newSchedule,
     });
 
-    // FIXME:
+    // FIXME(see REMOVE_ME above):
     // renew lease as a heartbeat side effect of the worker
     // this is fine most of the time though (assuming worker execution is within the lease lifespan)
     this.logger.info(`Renew lease ${runId}`);
@@ -501,7 +514,6 @@ function prettyRun(run: Run) {
 
 function validateRunIntegrity(run: Run) {
   const logger = getLoggerByAddress(import.meta, "substantial");
-  console.log("AAA", prettyRun(run));
 
   let life = 0;
 

src/typegate/src/runtimes/substantial/common.ts

@@ -109,12 +109,16 @@ export class Interrupt extends Error {
   }
 }
 
+export function runHasStarted(run: Run) {
+  return run.operations.some(({ event }) => event.type == "Start");
+}
+
 export function runHasStopped(run: Run) {
   return run.operations.some(({ event }) => event.type == "Stop");
 }
 
 export function checkOperationHasBeenScheduled(run: Run, operation: Operation) {
   return run.operations.some(({ at, event }) =>
-    at == operation.at && event == operation.event
+    at == operation.at && event.type == operation.event.type
   );
 }

src/typegate/src/runtimes/substantial/deno_context.ts

@@ -34,7 +34,11 @@ export class Context {
 
   #appendOp(op: OperationEvent) {
     if (!runHasStopped(this.run)) {
-      console.log("Append context", op.type);
+      // console.log(
+      //   "Append context",
+      //   op.type,
+      //   this.run.operations.map((o) => o.event.type),
+      // );
       this.run.operations.push({ at: new Date().toJSON(), event: op });
     }
   }