fix: fuel handling (#2946)

crates/fluvio-smartengine/src/engine.rs

@@ -125,9 +125,9 @@ impl SmartModuleChainInstance {
             for instance in instances {
                 // pass raw inputs to transform instance
                 // each raw input may result in multiple records
-                let starting_fuel = self.store.get_current_fuel();
+                self.store.top_up_fuel();
                 let output = instance.process(next_input, &mut self.store)?;
-                let fuel_used = starting_fuel - self.store.get_current_fuel();
+                let fuel_used = self.store.get_used_fuel();
                 debug!(fuel_used, "fuel used");
                 metric.add_fuel_used(fuel_used);
 
@@ -140,9 +140,9 @@ impl SmartModuleChainInstance {
                 }
             }
 
-            let starting_fuel = self.store.get_current_fuel();
+            self.store.top_up_fuel();
             let output = last.process(next_input, &mut self.store)?;
-            let fuel_used = starting_fuel - self.store.get_current_fuel();
+            let fuel_used = self.store.get_used_fuel();
             debug!(fuel_used, "fuel used");
             metric.add_fuel_used(fuel_used);
             let records_out = output.successes.len();
@@ -300,6 +300,8 @@ mod chaining_test {
         assert_eq!(output.successes[0].value.as_ref(), b"APPLE");
         assert_eq!(output.successes[1].value.as_ref(), b"BANANA");
         assert!(metrics.fuel_used() > 0);
+        chain.store.top_up_fuel();
+        assert_eq!(chain.store.get_used_fuel(), 0);
     }
 
     const SM_AGGEGRATE: &str = "fluvio_smartmodule_aggregate";
@@ -371,6 +373,9 @@ mod chaining_test {
         let mut chain = chain_builder
             .initialize(&engine)
             .expect("failed to build chain");
+
+        assert_eq!(chain.store.get_used_fuel(), 0);
+
         let record = vec![Record::new("input")];
         let input = SmartModuleInput::try_from(record).expect("valid input record");
         let metrics = SmartModuleChainMetrics::default();

crates/fluvio-smartengine/src/instance.rs

@@ -52,7 +52,6 @@ impl SmartModuleInstance {
         input: SmartModuleInput,
         store: &mut WasmState,
     ) -> Result<SmartModuleOutput> {
-        store.top_up_fuel();
         self.transform.process(input, &mut self.ctx, store)
     }
 

crates/fluvio-smartengine/src/state.rs

@@ -6,7 +6,10 @@ use wasmtime::{
     StoreContextMut,
 };
 
-const DEFAULT_FUEL: u64 = i64::MAX as u64 - 1;
+// DO NOT INCREASE THIS VALUE HIGHER THAN i64::MAX / 2.
+// WASMTIME keeps fuel as i64 and has some strange behavior with `add_fuel` if trying to top fuel
+// up to a values close to i64:MAX
+const DEFAULT_FUEL: u64 = i64::MAX as u64 / 2;
 
 #[cfg(not(feature = "wasi"))]
 pub type WasmState = WasmStore<()>;
@@ -40,10 +43,10 @@ impl WasmState {
         }
     }
 
-    // Get current fuel
-    pub fn get_current_fuel(&mut self) -> u64 {
+    // Get amount of fuel used since last top up
+    pub fn get_used_fuel(&mut self) -> u64 {
         if let Ok(current_fuel) = self.0.consume_fuel(0) {
-            current_fuel
+            max(DEFAULT_FUEL - current_fuel, 0)
         } else {
             0
         }