Two phase commit for mobile packet verifier

mobile_packet_verifier/Cargo.toml

@@ -21,6 +21,7 @@ helium-crypto = {workspace = true, features = ["sqlx-postgres", "multisig", "sol
 metrics = {workspace = true}
 poc-metrics = {path = "../metrics"}
 prost = {workspace = true}
+solana-sdk = {workspace = true}
 serde = {workspace = true}
 sqlx = {workspace = true}
 solana = {path = "../solana"}

mobile_packet_verifier/migrations/7_two_phase_commit.sql

@@ -0,0 +1,29 @@
+CREATE TABLE pending_txns AS (
+       signature TEXT PRIMARY KEY,
+       pub_key TEXT NOT NULL,
+       payer TEXT NOT NULL,
+       num_dcs BIGINT NOT NULL,
+       uploaded_bytes BIGINT NOT NULL,
+       downloaded_bytes BIGINT NOT NULL,
+       rewardable_bytes BIGINT NOT NULL,
+       first_timestamp TIMESTAMPTZ NOT NULL,
+       last_timestmap TIMESTAMPTZ NOT NULL
+);
+
+CREATE TABLE data_transfer_sessions_by_row (
+       pub_key TEXT NOT NULL,
+       payer TEXT NOT NULL,
+       uploaded_bytes BIGINT NOT NULL,
+       downloaded_bytes BIGINT NOT NULL,
+       rewardable_bytes BIGINT NOT NULL,
+       session_timestamp TIMESTAMPTZ NOT NULL,
+       PRIMARY KEY(pub_key, payer, session_timestamp)
+);
+
+INSERT INTO data_transfer_sessions_by_row
+       (pub_key, payer, uploaded_bytes, downloaded_bytes, rewardable_bytes, session_timestamp)
+SELECT pub_key, payer, uploaded_bytes, downloaded_bytes, rewardable_bytes, last_timestamp as session_timestamp FROM data_transfer_sessions;
+
+ALTER TABLE data_transfer_sessions RENAME TO old_data_transfer_sessions;
+ALTER TABLE data_transfer_sessions_by_row to data_transfer_sessions;
+DROP TABLE data_transfer_sessions;

mobile_packet_verifier/src/accumulate.rs

@@ -49,17 +49,11 @@ pub async fn accumulate_sessions(
         let event = report.report.data_transfer_usage;
         sqlx::query(
             r#"
-            INSERT INTO data_transfer_sessions (pub_key, payer, uploaded_bytes, downloaded_bytes, rewardable_bytes, first_timestamp, last_timestamp)
-            VALUES ($1, $2, $3, $4, $5, $6, $6)
-            ON CONFLICT (pub_key, payer) DO UPDATE SET
-            uploaded_bytes = data_transfer_sessions.uploaded_bytes + EXCLUDED.uploaded_bytes,
-            downloaded_bytes = data_transfer_sessions.downloaded_bytes + EXCLUDED.downloaded_bytes,
-            rewardable_bytes = data_transfer_sessions.rewardable_bytes + EXCLUDED.rewardable_bytes,
-            last_timestamp = GREATEST(data_transfer_sessions.last_timestamp, EXCLUDED.last_timestamp)
+            INSERT INTO data_transfer_sessions (pub_key, payer, uploaded_bytes, downloaded_bytes, rewardable_bytes, session_timestamp) VALUES ($1, $2, $3, $4, $5, $6);
             "#
         )
             .bind(event.pub_key)
-            .bind(event.payer)
+            .bind(&event.payer)
             .bind(event.upload_bytes as i64)
             .bind(event.download_bytes as i64)
             .bind(report.report.rewardable_bytes as i64)