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)