feat: enable cohort to pass on_commit messages for messenger

cohort_banking_initiator_js/src/banking-app.ts

@@ -21,7 +21,7 @@ export class BankingApp {
         private pond: Pond,
         private database: Pool,
         private queue: BroadcastChannel,
-        private onFinishListener: (appRef: BankingApp) => any) {}
+        private onFinishListener: (appRef: BankingApp) => any) { }
 
     async init() {
         this.initiator = await Initiator.init(sdkConfig)
@@ -133,6 +133,22 @@ export class BankingApp {
                 writeset: [tx.from, tx.to],
                 readvers: state.items.map(i => i.version),
                 statemaps: [{ "TRANSFER": tx }],
+                onCommit: {
+                    publish: {
+                        kafka: [
+                            {
+                                topic: "test.transfer.feedback.js",
+                                value: {
+                                    "from_account": tx.from,
+                                    "to_account": tx.to,
+                                    "amount": tx.amount
+                                }
+                            },
+                        ]
+                    }
+                }
+
+                // TODO: GK - Add on commit action
             },
             snapshot: state.snapshotVersion,
             timeoutMs: 0,
@@ -143,14 +159,16 @@ export class BankingApp {
         let cnn: PoolClient
         try {
             cnn = await this.database.connect()
-            const result = await cnn.query({ name: "get-state", text:
-                `SELECT
+            const result = await cnn.query({
+                name: "get-state", text:
+                    `SELECT
                     ba."number" as "id", ba."version" as "version", cs."version" AS snapshot_version
                 FROM
                     bank_accounts ba, cohort_snapshot cs
                 WHERE
                     ba."number" = $1 OR ba."number" = $2`,
-                values: [tx.from, tx.to] }
+                values: [tx.from, tx.to]
+            }
             )
 
             if (result.rowCount != 2) {
@@ -163,7 +181,7 @@ export class BankingApp {
         } catch (e) {
             // This print here is important, without it the original reason is lost when using NAPI 2.10.
             logger.error("BankingApp.loadState(): %s", e)
-            throw new Error(`Unable to load state for tx: ${ JSON.stringify(tx) }. Reason: ${e.message}`, { cause: e })
+            throw new Error(`Unable to load state for tx: ${JSON.stringify(tx)}. Reason: ${e.message}`, { cause: e })
         } finally {
             cnn?.release()
         }
@@ -219,7 +237,7 @@ export class BankingApp {
         } catch (e) {
             // This print here is important, without it the original reason is lost when using NAPI 2.10.
             logger.error("BankingApp.installOutOfOrder(): %s", e)
-            throw new Error(`Unable to complete out of order installation of tx: ${ JSON.stringify(tx) }`, { cause: e })
+            throw new Error(`Unable to complete out of order installation of tx: ${JSON.stringify(tx)}`, { cause: e })
         } finally {
             cnn?.release()
         }

examples/agent_client/examples/agent_client.rs

@@ -297,6 +297,7 @@ impl Generator<CertificationRequest> for RequestGenerator {
             snapshot: 5,
             writeset: Vec::from(["3".to_string()]),
             statemap: None,
+            on_commit: None,
         };
 
         CertificationRequest {

packages/cohort_banking/src/app.rs

@@ -14,6 +14,7 @@ use opentelemetry_api::{
     global,
     metrics::{Counter, Unit},
 };
+use serde_json::json;
 use talos_agent::messaging::api::Decision;
 
 use crate::{
@@ -64,19 +65,35 @@ impl BankingApp {
 #[async_trait]
 impl Handler<TransferRequest> for BankingApp {
     async fn handle(&self, request: TransferRequest) -> Result<(), String> {
-        log::debug!("processig new banking transfer request: {:?}", request);
+        log::debug!("processing new banking transfer request: {:?}", request);
 
         let statemap = vec![HashMap::from([(
             BusinessActionType::TRANSFER.to_string(),
             TransferRequest::new(request.from.clone(), request.to.clone(), request.amount).json(),
         )])];
 
+        let on_commit_value = json!({
+            "publish": {
+                "kafka": [
+                    {
+                        "topic": "test.transfer.feedback",
+                        "value": {
+                            "from_account": request.from,
+                            "to_account": request.to,
+                            "amount": request.amount
+                        }
+                    },
+                ],
+            }
+        });
+
         let certification_request = CertificationRequest {
             timeout_ms: 0,
             candidate: CandidateData {
                 readset: vec![request.from.clone(), request.to.clone()],
                 writeset: vec![request.from, request.to],
                 statemap: Some(statemap),
+                on_commit: Some(on_commit_value),
             },
         };
 
@@ -85,6 +102,7 @@ impl Handler<TransferRequest> for BankingApp {
             database: Arc::clone(&self.database),
             single_query_strategy,
         };
+
         let request_payload_callback = || state_provider.get_certification_candidate(certification_request.clone());
 
         let oo_inst = OutOfOrderInstallerImpl {

packages/cohort_banking/src/callbacks/certification_candidate_provider.rs

@@ -155,6 +155,7 @@ impl CertificationCandidateProviderImpl {
             writeset: request.candidate.writeset,
             statemaps: request.candidate.statemap,
             readvers: state.items.into_iter().map(|x| x.version).collect(),
+            on_commit: request.candidate.on_commit.map(|x| x.into()),
         };
 
         Ok(CertificationCandidateCallbackResponse::Proceed(CertificationRequestPayload {

packages/cohort_banking/src/examples_support/queue_processor.rs

@@ -22,7 +22,6 @@ impl QueueProcessor {
         threads: u64,
         item_handler: Arc<H>,
     ) -> Vec<JoinHandle<()>> {
-        let item_handler = Arc::new(item_handler);
         let mut tasks = Vec::<JoinHandle<()>>::new();
 
         for thread_number in 1..=threads {

packages/cohort_banking/src/model/requests.rs

@@ -6,7 +6,8 @@ pub struct CandidateData {
     pub readset: Vec<String>,
     pub writeset: Vec<String>,
     pub statemap: Option<Vec<HashMap<String, Value>>>,
-    // The "snapshot" is intentionally messing here. We will compute it ourselves before feeding this data to Talos
+    // The "snapshot" is intentionally missing here. We will compute it ourselves before feeding this data to Talos
+    pub on_commit: Option<Value>,
 }
 
 #[derive(Clone)]

packages/cohort_sdk/src/cohort.rs

@@ -457,6 +457,7 @@ impl Cohort {
                 writeset: request.candidate.writeset,
                 readvers,
                 snapshot,
+                on_commit: request.candidate.on_commit,
             },
             timeout: if request.timeout_ms > 0 {
                 Some(Duration::from_millis(request.timeout_ms))

packages/cohort_sdk/src/model/callback.rs

@@ -22,6 +22,7 @@ pub struct CertificationCandidate {
     pub writeset: Vec<String>,
     pub readvers: Vec<u64>,
     pub statemaps: Option<Vec<HashMap<String, Value>>>,
+    pub on_commit: Option<Box<Value>>,
 }
 
 #[derive(Debug, Clone)]

packages/cohort_sdk_js/src/initiator/mod.rs

@@ -73,6 +73,7 @@ pub struct JsCertificationCandidate {
     pub writeset: Vec<String>,
     pub readvers: Vec<i64>,
     pub statemaps: Option<Vec<HashMap<String, Value>>>,
+    pub on_commit: Option<Value>,
 }
 
 impl From<JsCertificationCandidate> for CertificationCandidate {
@@ -82,6 +83,7 @@ impl From<JsCertificationCandidate> for CertificationCandidate {
             writeset: val.writeset,
             readvers: val.readvers.iter().map(|v| *v as u64).collect(),
             statemaps: val.statemaps,
+            on_commit: val.on_commit.map(Box::new),
         }
     }
 }

packages/talos_agent/src/agent/core.rs

@@ -246,6 +246,7 @@ mod tests {
             snapshot: 1_u64,
             writeset: Vec::<String>::new(),
             statemap: None,
+            on_commit: None,
         }
     }
 

packages/talos_agent/src/agent/errors.rs

@@ -100,6 +100,7 @@ mod tests {
                     snapshot: 0,
                     writeset: vec![String::from("1"), String::from("2"), String::from("3")],
                     statemap: None,
+                    on_commit: None,
                 },
                 timeout: Some(Duration::from_secs(1)),
             },

packages/talos_agent/src/agent/state_manager.rs

@@ -337,6 +337,7 @@ mod tests {
                     snapshot: 1_u64,
                     writeset: Vec::<String>::new(),
                     statemap: None,
+                    on_commit: None,
                 },
             },
             Arc::new(Box::new(tx_answer)),
@@ -668,6 +669,7 @@ mod tests {
                     snapshot: 0,
                     writeset: Vec::<String>::new(),
                     statemap: None,
+                    on_commit: None,
                 },
             },
         };

packages/talos_agent/src/api.rs

@@ -21,6 +21,7 @@ pub struct CandidateData {
     pub snapshot: u64,
     pub writeset: Vec<String>,
     pub statemap: Option<StateMap>,
+    pub on_commit: Option<Box<Value>>,
 }
 
 /// The data input from client to agent

packages/talos_agent/src/messaging/api.rs

@@ -2,6 +2,7 @@ use crate::api::{CandidateData, StateMap, TalosType};
 use crate::messaging::errors::MessagingError;
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
+use serde_json::Value;
 use strum::{Display, EnumString};
 
 pub static HEADER_MESSAGE_TYPE: &str = "messageType";
@@ -27,6 +28,8 @@ pub struct CandidateMessage {
     pub writeset: Vec<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub statemap: Option<StateMap>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub on_commit: Option<Box<Value>>,
     pub published_at: i128,
 }
 
@@ -41,6 +44,7 @@ impl CandidateMessage {
             snapshot: candidate.snapshot,
             writeset: candidate.writeset,
             statemap: candidate.statemap,
+            on_commit: candidate.on_commit,
             published_at,
         }
     }
@@ -139,6 +143,7 @@ mod tests {
                 snapshot: 1_u64,
                 writeset: vec!["1".to_string()],
                 statemap: None,
+                on_commit: None,
             },
             0,
         );

packages/talos_agent/src/messaging/kafka.rs

@@ -332,6 +332,7 @@ mod tests_publisher {
             snapshot: 2_u64,
             writeset: vec!["1".to_string()],
             statemap: None,
+            on_commit: None,
             published_at: 0,
         })
         .unwrap();