feat: even more sync progress, including verified working testing scripts!

Author: ErikBjare

aw-client-rust/src/lib.rs

@@ -29,7 +29,7 @@ impl AwClient {
     pub fn new(ip: &str, port: &str, name: &str) -> AwClient {
         let baseurl = format!("http://{}:{}", ip, port);
         let client = reqwest::blocking::Client::builder()
-            .timeout(std::time::Duration::from_secs(60))
+            .timeout(std::time::Duration::from_secs(120))
             .build()
             .unwrap();
         let hostname = gethostname::gethostname().into_string().unwrap();
@@ -52,9 +52,18 @@ impl AwClient {
         self.client.get(&url).send()?.json()
     }
 
-    pub fn create_bucket(&self, bucketname: &str, buckettype: &str) -> Result<(), reqwest::Error> {
-        let url = format!("{}/api/0/buckets/{}", self.baseurl, bucketname);
-        let data = Bucket {
+    pub fn create_bucket(&self, bucket: &Bucket) -> Result<(), reqwest::Error> {
+        let url = format!("{}/api/0/buckets/{}", self.baseurl, bucket.id);
+        self.client.post(&url).json(bucket).send()?;
+        Ok(())
+    }
+
+    pub fn create_bucket_simple(
+        &self,
+        bucketname: &str,
+        buckettype: &str,
+    ) -> Result<(), reqwest::Error> {
+        let bucket = Bucket {
             bid: None,
             id: bucketname.to_string(),
             client: self.name.clone(),
@@ -66,8 +75,7 @@ impl AwClient {
             created: None,
             last_updated: None,
         };
-        self.client.post(&url).json(&data).send()?;
-        Ok(())
+        self.create_bucket(&bucket)
     }
 
     pub fn delete_bucket(&self, bucketname: &str) -> Result<(), reqwest::Error> {
@@ -111,6 +119,16 @@ impl AwClient {
         Ok(())
     }
 
+    pub fn insert_events(
+        &self,
+        bucketname: &str,
+        events: Vec<Event>,
+    ) -> Result<(), reqwest::Error> {
+        let url = format!("{}/api/0/buckets/{}/events", self.baseurl, bucketname);
+        self.client.post(&url).json(&events).send()?;
+        Ok(())
+    }
+
     pub fn heartbeat(
         &self,
         bucketname: &str,

aw-client-rust/tests/test.rs

@@ -76,7 +76,9 @@ mod test {
 
         let bucketname = format!("aw-client-rust-test_{}", client.hostname);
         let buckettype = "test-type";
-        client.create_bucket(&bucketname, &buckettype).unwrap();
+        client
+            .create_bucket_simple(&bucketname, &buckettype)
+            .unwrap();
 
         let bucket = client.get_bucket(&bucketname).unwrap();
         assert!(bucket.id == bucketname);

aw-sync/src/accessmethod.rs

@@ -19,7 +19,7 @@ pub trait AccessMethod: std::fmt::Debug {
         end: Option<DateTime<Utc>>,
         limit: Option<u64>,
     ) -> Result<Vec<Event>, String>;
-    fn insert_events(&self, bucket_id: &str, events: Vec<Event>) -> Result<Vec<Event>, String>;
+    fn insert_events(&self, bucket_id: &str, events: Vec<Event>) -> Result<(), String>;
     fn get_event_count(&self, bucket_id: &str) -> Result<i64, String>;
     fn heartbeat(&self, bucket_id: &str, event: Event, duration: f64) -> Result<(), String>;
     fn close(&self);
@@ -30,10 +30,10 @@ impl AccessMethod for Datastore {
         Ok(self.get_buckets().unwrap())
     }
     fn get_bucket(&self, bucket_id: &str) -> Result<Bucket, DatastoreError> {
-        self.get_bucket(bucket_id)
+        Datastore::get_bucket(self, bucket_id)
     }
     fn create_bucket(&self, bucket: &Bucket) -> Result<(), DatastoreError> {
-        self.create_bucket(bucket)?;
+        Datastore::create_bucket(self, bucket)?;
         self.force_commit().unwrap();
         Ok(())
     }
@@ -44,23 +44,23 @@ impl AccessMethod for Datastore {
         end: Option<DateTime<Utc>>,
         limit: Option<u64>,
     ) -> Result<Vec<Event>, String> {
-        Ok(self.get_events(bucket_id, start, end, limit).unwrap())
+        Ok(Datastore::get_events(self, bucket_id, start, end, limit).unwrap())
     }
     fn heartbeat(&self, bucket_id: &str, event: Event, duration: f64) -> Result<(), String> {
-        self.heartbeat(bucket_id, event, duration).unwrap();
+        Datastore::heartbeat(self, bucket_id, event, duration).unwrap();
         self.force_commit().unwrap();
         Ok(())
     }
-    fn insert_events(&self, bucket_id: &str, events: Vec<Event>) -> Result<Vec<Event>, String> {
-        let res = self.insert_events(bucket_id, &events[..]).unwrap();
+    fn insert_events(&self, bucket_id: &str, events: Vec<Event>) -> Result<(), String> {
+        Datastore::insert_events(self, bucket_id, &events[..]).unwrap();
         self.force_commit().unwrap();
-        Ok(res)
+        Ok(())
     }
     fn get_event_count(&self, bucket_id: &str) -> Result<i64, String> {
-        Ok(self.get_event_count(bucket_id, None, None).unwrap())
+        Ok(Datastore::get_event_count(self, bucket_id, None, None).unwrap())
     }
     fn close(&self) {
-        self.close();
+        Datastore::close(self);
     }
 }
 
@@ -92,18 +92,15 @@ impl AccessMethod for AwClient {
     ) -> Result<Vec<Event>, String> {
         Ok(self.get_events(bucket_id, start, end, limit).unwrap())
     }
-    fn insert_events(&self, _bucket_id: &str, _events: Vec<Event>) -> Result<Vec<Event>, String> {
-        //Ok(self.insert_events(bucket_id, &events[..]).unwrap())
-        Err("Not implemented".to_string())
+    fn insert_events(&self, bucket_id: &str, events: Vec<Event>) -> Result<(), String> {
+        AwClient::insert_events(self, bucket_id, events).map_err(|e| e.to_string())
     }
     fn get_event_count(&self, bucket_id: &str) -> Result<i64, String> {
         Ok(self.get_event_count(bucket_id).unwrap())
     }
     fn create_bucket(&self, bucket: &Bucket) -> Result<(), DatastoreError> {
-        self.create_bucket(bucket.id.as_str(), bucket._type.as_str())
-            .unwrap();
+        AwClient::create_bucket(self, bucket).unwrap();
         Ok(())
-        //Err(DatastoreError::InternalError("Not implemented".to_string()))
     }
     fn heartbeat(&self, bucket_id: &str, event: Event, duration: f64) -> Result<(), String> {
         self.heartbeat(bucket_id, &event, duration)

aw-sync/src/sync.rs

@@ -317,6 +317,10 @@ fn sync_one(
     } else {
         info!("   + Starting from beginning");
     }
+
+    // Fetch events
+    // Unset ID on events, as they are not globally unique
+    // TODO: Fetch at most ~5,000 events at a time (or so, to avoid timeout from huge buckets)
     let mut events: Vec<Event> = ds_from
         .get_events(bucket_from.id.as_str(), resume_sync_at, None, None)
         .unwrap()
@@ -329,13 +333,47 @@ fn sync_one(
         .collect();
 
     // Sort ascending
+    // FIXME: What happens here if two events have the same timestamp?
     events.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
-    //info!("{:?}", events);
 
     // TODO: Do bulk insert using insert_events instead? (for performance)
-    for event in events {
-        //print!("\r{}", event.timestamp);
-        ds_to.heartbeat(bucket_to.id.as_str(), event, 0.0).unwrap();
+    //       Client-side heartbeat queueing should keep things somewhat performant though?
+    // NOTE: First event needs to be inserted with heartbeat, to ensure appropriate
+    // merging/updating of pulsed events.
+    let events_total = events.len();
+    let mut events_sent = 0;
+    let mut events_iter = events.into_iter();
+    if let Some(e) = events_iter.next() {
+        ds_to.heartbeat(bucket_to.id.as_str(), e, 0.0).unwrap();
+        events_sent += 1;
+    }
+
+    const BATCH_SIZE: usize = 5000;
+    if BATCH_SIZE == 1 {
+        for event in events_iter {
+            print!("{} ({}/{})\r", &event.timestamp, events_sent, events_total);
+            ds_to.heartbeat(bucket_to.id.as_str(), event, 0.0).unwrap();
+            events_sent += 1;
+        }
+    } else {
+        let mut batch_events = Vec::with_capacity(BATCH_SIZE);
+        for e in events_iter {
+            print!("{} ({}/{})\r", e.timestamp, events_sent, events_total);
+            batch_events.push(e);
+            events_sent += 1;
+            if batch_events.len() >= BATCH_SIZE {
+                ds_to
+                    .insert_events(bucket_to.id.as_str(), batch_events.clone())
+                    .unwrap();
+                batch_events.clear();
+            }
+        }
+
+        if !batch_events.is_empty() {
+            ds_to
+                .insert_events(bucket_to.id.as_str(), batch_events)
+                .unwrap();
+        }
     }
 
     let eventcount_to_new = ds_to.get_event_count(bucket_to.id.as_str()).unwrap();

aw-sync/test-import-sync.sh

@@ -0,0 +1,36 @@
+#!/bin/bash
+
+# get script path
+SCRIPTPATH="$( cd "$(dirname "$0")" ; pwd -P )"
+pushd $SCRIPTPATH
+
+# port used for testing instance
+PORT=5667
+
+# if server already running on port 5667, don't start again
+if [ "$(lsof -i:$PORT -sTCP:LISTEN -t)" ]; then
+    echo "ActivityWatch server already running on port $PORT, using that."
+else
+    # Set up an isolated ActivityWatch instance
+    ./test-server.sh $PORT &
+fi
+
+
+sleep 1;
+SYNCROOTDIR="$HOME/ActivityWatchSync"
+
+# For each host in the sync directory, pull the data from each database file using aw-sync
+for host in $(ls $SYNCROOTDIR); do
+    SYNCDIR="$SYNCROOTDIR/$host"
+    for db in $(ls $SYNCDIR/*/*.db); do
+        AWSYNCPARAMS="--port $PORT --sync-dir $SYNCDIR"
+        BUCKETS="aw-watcher-window_$host,aw-watcher-afk_$host"
+
+        echo "Syncing $db to $host"
+        cargo run --bin aw-sync -- $AWSYNCPARAMS sync --mode pull --buckets $BUCKETS
+    done
+done
+
+# kill aw-server-rust
+#kill %1
+fg

aw-sync/test-server.sh

@@ -0,0 +1,13 @@
+#!/bin/bash
+
+# port and db used for testing instance
+
+# If port already set, use that, otherwise, use 5667
+PORT=${PORT:-5667}
+
+DBPATH=/tmp/aw-server-rust-sync-testing/
+mkdir -p $DBPATH
+
+# Set up an isolated ActivityWatch instance
+pushd ..
+cargo run --bin aw-server -- --testing --port $PORT --dbpath $DBPATH/data.db --no-legacy-import