Merge pull request #529 from AmbireTech/improve-sentry-event-submission-performance

Improve sentry event submission performance

Author: elpiel

Cargo.lock

@@ -149,14 +149,14 @@ mod logger {
 pub static DUMMY_VALIDATOR_LEADER: Lazy<ValidatorDesc> = Lazy::new(|| ValidatorDesc {
     id: IDS[&LEADER],
     url: "http://localhost:8005".to_string(),
-    fee: UnifiedNum::from_whole(0.02),
+    fee: UnifiedNum::from_whole(0.03),
     fee_addr: None,
 });
 
 pub static DUMMY_VALIDATOR_FOLLOWER: Lazy<ValidatorDesc> = Lazy::new(|| ValidatorDesc {
     id: IDS[&FOLLOWER],
     url: "http://localhost:8006".to_string(),
-    fee: UnifiedNum::from_whole(0.03),
+    fee: UnifiedNum::from_whole(0.02),
     fee_addr: None,
 });
 

primitives/src/test_util.rs

@@ -52,7 +52,7 @@ args = [
   "-t3",
   "-c100",
   "-d30s",
-  "-R2000",
+  "-R3000",
   "--latency",
   "http://127.0.0.1:8005/v5/campaign",
 ]

sentry/Makefile.toml

@@ -1,3 +1,5 @@
+-- Multiple campaigns with a single event in the request
+--
 -- This script will submit events for 3 campaigns
 -- The 3 campaigns can be found in `primitives::test_util`
 wrk.method = "POST"
@@ -7,7 +9,6 @@ wrk.headers["Content-Type"] = "application/json"
 -- uses the DUMMY_AUTH[CREATOR] token
 -- wrk.headers["authorization"] = "Bearer AUTH_awesomeCreator:chain_id:1337"
 
-
 init = function(args)
     local r = {}
 

sentry/benchmark/multiple_benchmark.lua

@@ -0,0 +1,34 @@
+-- Multiple events to multiple campaigns
+--
+-- This script will submit events for 3 campaigns
+-- The 3 campaigns can be found in `primitives::test_util`
+-- Each requests consist of 2 events - IMPRESSION & CLICK
+-- and each event has differnet publihser (PUBLIHSER & PUBLISHER_2) as well as different AdUnit & AdSlot
+-- The same events are used for all campaigns.
+wrk.method = "POST"
+-- uses the PUBLISHER (for IMPRESSION) & PUBLISHER_2 (for CLICK) address
+wrk.body = "{ \"events\": [ {\"type\": \"IMPRESSION\", \"publisher\": \"0xE882ebF439207a70dDcCb39E13CA8506c9F45fD9\", \"adUnit\": \"Qmasg8FrbuSQpjFu3kRnZF9beg8rEBFrqgi1uXDRwCbX5f\", \"adSlot\": \"QmcUVX7fvoLMM93uN2bD3wGTH8MXSxeL8hojYfL2Lhp7mR\"} ] }"
+wrk.headers["Content-Type"] = "application/json"
+-- uses the DUMMY_AUTH[CREATOR] token
+-- wrk.headers["authorization"] = "Bearer AUTH_awesomeCreator:chain_id:1337"
+
+
+init = function(args)
+    local r = {}
+
+    -- with 2 different publishers (PUBLISHER, PUBLISHER_2), different AdUnits & different AdSlots
+    local two_events_body = "{ \"events\": [ {\"type\": \"IMPRESSION\", \"publisher\": \"0xE882ebF439207a70dDcCb39E13CA8506c9F45fD9\", \"adUnit\": \"Qmasg8FrbuSQpjFu3kRnZF9beg8rEBFrqgi1uXDRwCbX5f\", \"adSlot\": \"QmcUVX7fvoLMM93uN2bD3wGTH8MXSxeL8hojYfL2Lhp7mR\"}, {\"type\": \"CLICK\", \"publisher\": \"0x0e880972A4b216906F05D67EeaaF55d16B5EE4F1\", \"adUnit\": \"QmQnu8zrHsuVvnTJsEgDHYA8c1MmRL7YLiMD8uzDUJKcNq\", \"adSlot\": \"QmYYBULc9QDEaDr8HAXvVWHDmFfL2GvyumYRr1g4ERBC96\"} ] }"
+
+    -- Campaign 1
+    r[1] = wrk.format(nil, "/v5/campaign/0x936da01f9abd4d9d80c702af85c822a8/events", nil, two_events_body)
+    -- Campaign 2
+    r[2] = wrk.format(nil, "/v5/campaign/0x127b98248f4e4b73af409d10f62daeaa/events", nil, two_events_body)
+    -- Campaign 3
+    r[3] = wrk.format(nil, "/v5/campaign/0xa78f3492481b41a688488a7aa1ff17df/events", nil, two_events_body)
+
+    req = table.concat(r)
+end
+
+request = function()
+    return req
+ end

sentry/benchmark/multiple_events.lua

@@ -72,7 +72,7 @@ CREATE TABLE accounting (
     side AccountingSide NOT NULL,
     "address" varchar(42) NOT NULL,
     amount bigint NOT NULL,
-    updated timestamp(2) with time zone DEFAULT NULL NULL,
+    updated timestamp(2) with time zone DEFAULT NULL,
     created timestamp(2) with time zone NOT NULL,
     -- Do not rename the Primary key constraint (`accounting_pkey`)!
     PRIMARY KEY (channel_id, side, "address"),

sentry/migrations/20190806011140_initial-tables/up.sql

@@ -42,6 +42,7 @@ pub async fn check_access(
         return Err(Error::ForbiddenReferrer);
     }
 
+    // Used when `Campaign::event_submission` is `None`
     let default_rules = [
         Rule {
             uids: Some(vec![campaign.creator.to_string()]),
@@ -84,7 +85,7 @@ pub async fn check_access(
 }
 
 async fn apply_rule(
-    redis: MultiplexedConnection,
+    mut redis: MultiplexedConnection,
     rule: &Rule,
     events: &[Event],
     campaign: &Campaign,
@@ -116,7 +117,7 @@ async fn apply_rule(
 
             if redis::cmd("EXISTS")
                 .arg(&key)
-                .query_async::<_, i8>(&mut redis.clone())
+                .query_async::<_, i8>(&mut redis)
                 .await
                 .map(|exists| exists == 1)
                 .map_err(|error| format!("{}", error))?
@@ -129,7 +130,7 @@ async fn apply_rule(
                 .arg(&key)
                 .arg(seconds as i32)
                 .arg("1")
-                .query_async::<_, ()>(&mut redis.clone())
+                .query_async::<_, ()>(&mut redis)
                 .await
                 .map_err(|error| format!("{}", error))
         }

sentry/src/access.rs

@@ -2,6 +2,7 @@ use crate::{
     db::{analytics::update_analytics, DbPool, PoolError},
     Session,
 };
+use futures::future::join_all;
 use primitives::{
     analytics::OperatingSystem,
     sentry::{DateHour, Event, UpdateAnalytics},
@@ -14,7 +15,7 @@ pub async fn record(
     pool: &DbPool,
     campaign_context: &ChainOf<Campaign>,
     session: &Session,
-    events_with_payouts: Vec<(Event, Address, UnifiedNum)>,
+    events_with_payouts: &[(Event, Address, UnifiedNum)],
 ) -> Result<(), PoolError> {
     let os_name = session
         .os
@@ -65,9 +66,9 @@ pub async fn record(
         };
 
         batch_update
-            .entry(event)
+            .entry(event.clone())
             .and_modify(|analytics| {
-                analytics.amount_to_add += &payout_amount;
+                analytics.amount_to_add += payout_amount;
                 analytics.count_to_add += 1;
             })
             .or_insert_with(|| UpdateAnalytics {
@@ -83,14 +84,24 @@ pub async fn record(
                 os_name: os_name.clone(),
                 chain_id: campaign_context.chain.chain_id,
                 event_type,
-                amount_to_add: payout_amount,
+                amount_to_add: *payout_amount,
                 count_to_add: 1,
             });
     }
 
-    for (_event, update) in batch_update.into_iter() {
-        update_analytics(pool, update).await?;
-    }
+    let batch_futures = join_all(
+        batch_update
+            .into_iter()
+            .map(|(_event, update)| update_analytics(pool, update)),
+    );
+
+    // execute the batched futures, collect the result afterwards,
+    // in order execute all futures first and then return an error if occurred
+    batch_futures
+        .await
+        .into_iter()
+        .collect::<Result<Vec<_>, _>>()?;
+
     Ok(())
 }
 
@@ -192,7 +203,7 @@ mod test {
         let channel_context = channel_chain.with_channel(dummy_channel);
         let campaign_context = channel_context.clone().with(campaign);
 
-        record(&app.pool, &campaign_context, &session, input_events.clone())
+        record(&app.pool, &campaign_context, &session, &input_events)
             .await
             .expect("should record");
 
@@ -256,7 +267,7 @@ mod test {
             .expect("Channel token should be whitelisted in config!");
         let channel_context = channel_chain.with_channel(dummy_channel);
         let campaign_context = channel_context.clone().with(campaign);
-        record(&app.pool, &campaign_context, &session, input_events.clone())
+        record(&app.pool, &campaign_context, &session, &input_events)
             .await
             .expect("should record");
 

sentry/src/analytics.rs

@@ -1,4 +1,7 @@
+use std::fmt;
+
 use chrono::{DateTime, Utc};
+use futures::future::{join, join_all};
 use primitives::{
     balances::{Balances, CheckedState},
     Address, ChannelId, UnifiedNum,
@@ -11,7 +14,7 @@ use tokio_postgres::{
 use super::{DbPool, PoolError};
 use thiserror::Error;
 
-static UPDATE_ACCOUNTING_STATEMENT: &str = "INSERT INTO accounting(channel_id, side, address, amount, updated, created) VALUES($1, $2, $3, $4, $5, $6) ON CONFLICT ON CONSTRAINT accounting_pkey DO UPDATE SET amount = accounting.amount + $4, updated = $6 WHERE accounting.channel_id = $1 AND accounting.side = $2 AND accounting.address = $3 RETURNING channel_id, side, address, amount, updated, created";
+static UPDATE_ACCOUNTING_STATEMENT: &str = "INSERT INTO accounting(channel_id, side, address, amount, updated, created) VALUES($1, $2, $3, $4, NULL, NOW()) ON CONFLICT ON CONSTRAINT accounting_pkey DO UPDATE SET amount = accounting.amount + EXCLUDED.amount, updated = NOW() WHERE accounting.channel_id = $1 AND accounting.side = $2 AND accounting.address = $3 RETURNING channel_id, side, address, amount, updated, created";
 
 #[derive(Debug, Error)]
 pub enum Error {
@@ -57,12 +60,16 @@ pub enum Side {
     Spender,
 }
 
-pub enum SpendError {
-    Pool(PoolError),
-    NoRecordsUpdated,
+impl fmt::Display for Side {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Side::Earner => write!(f, "Earner"),
+            Side::Spender => write!(f, "Spender"),
+        }
+    }
 }
 
-/// ```text
+/// ```sql
 /// SELECT channel_id, side, address, amount, updated, created FROM accounting WHERE channel_id = $1 AND address = $2 AND side = $3
 /// ```
 pub async fn get_accounting(
@@ -110,14 +117,8 @@ pub async fn update_accounting(
     let client = pool.get().await?;
     let statement = client.prepare(UPDATE_ACCOUNTING_STATEMENT).await?;
 
-    let now = Utc::now();
-    let updated: Option<DateTime<Utc>> = None;
-
     let row = client
-        .query_one(
-            &statement,
-            &[&channel_id, &side, &address, &amount, &updated, &now],
-        )
+        .query_one(&statement, &[&channel_id, &side, &address, &amount])
         .await?;
 
     Ok(Accounting::from(&row))
@@ -126,52 +127,63 @@ pub async fn update_accounting(
 /// `delta_balances` defines the Balances that need to be added to the spending or earnings of the `Accounting`s.
 /// It will **not** override the whole `Accounting` value
 /// Returns a tuple of `(Vec<Earners Accounting>, Vec<Spenders Accounting>)`
+///
+/// # Error
+///
+/// It will return an error if any of the updates fails but it would have updated the rest of them.
+///
+/// This way we ensure that even if a single or multiple Accounting updates fail,
+/// we will still pay out the rest of them.
 pub async fn spend_amount(
     pool: DbPool,
     channel_id: ChannelId,
     delta_balances: Balances<CheckedState>,
 ) -> Result<(Vec<Accounting>, Vec<Accounting>), PoolError> {
-    let client = pool.get().await?;
-
-    let statement = client.prepare(UPDATE_ACCOUNTING_STATEMENT).await?;
+    let client = &pool.get().await?;
 
-    let now = Utc::now();
-    let updated: Option<DateTime<Utc>> = None;
-
-    let (mut earners, mut spenders) = (vec![], vec![]);
+    let statement = client.prepare_cached(UPDATE_ACCOUNTING_STATEMENT).await?;
 
     // Earners
-    for (earner, amount) in delta_balances.earners {
-        let row = client
-            .query_one(
-                &statement,
-                &[&channel_id, &Side::Earner, &earner, &amount, &updated, &now],
-            )
-            .await?;
+    let earners_futures = delta_balances.earners.into_iter().map(|(earner, amount)| {
+        let statement = statement.clone();
 
-        earners.push(Accounting::from(&row))
-    }
+        async move {
+            client
+                .query_one(&statement, &[&channel_id, &Side::Earner, &earner, &amount])
+                .await
+                .map(|row| Accounting::from(&row))
+        }
+    });
 
     // Spenders
-    for (spender, amount) in delta_balances.spenders {
-        let row = client
-            .query_one(
-                &statement,
-                &[
-                    &channel_id,
-                    &Side::Spender,
-                    &spender,
-                    &amount,
-                    &updated,
-                    &now,
-                ],
-            )
-            .await?;
+    let spenders_futures = delta_balances
+        .spenders
+        .into_iter()
+        .map(|(spender, amount)| {
+            let statement = statement.clone();
+
+            async move {
+                client
+                    .query_one(
+                        &statement,
+                        &[&channel_id, &Side::Spender, &spender, &amount],
+                    )
+                    .await
+                    .map(|row| Accounting::from(&row))
+            }
+        });
 
-        spenders.push(Accounting::from(&row))
-    }
+    let earners = join_all(earners_futures);
+    let spenders = join_all(spenders_futures);
+
+    // collect all the Accounting updates into Vectors
+    let (earners, spenders) = join(earners, spenders).await;
 
-    Ok((earners, spenders))
+    // Return an error if any of the Accounting updates failed
+    Ok((
+        earners.into_iter().collect::<Result<_, _>>()?,
+        spenders.into_iter().collect::<Result<_, _>>()?,
+    ))
 }
 
 #[cfg(test)]
@@ -531,7 +543,7 @@ mod test {
                 earners_acc
                     .iter()
                     .find(|a| a.address == earner)
-                    .unwrap()
+                    .expect("Should find Accounting")
                     .clone(),
                 false,
             );
@@ -540,7 +552,7 @@ mod test {
                 earners_acc
                     .iter()
                     .find(|a| a.address == other_earner)
-                    .unwrap()
+                    .expect("Should find Accounting")
                     .clone(),
                 false,
             );
@@ -551,7 +563,7 @@ mod test {
                 spenders_acc
                     .iter()
                     .find(|a| a.address == spender)
-                    .unwrap()
+                    .expect("Should find Accounting")
                     .clone(),
                 false,
             );
@@ -560,7 +572,7 @@ mod test {
                 spenders_acc
                     .iter()
                     .find(|a| a.address == other_spender)
-                    .unwrap()
+                    .expect("Should find Accounting")
                     .clone(),
                 false,
             );