Reschedule pending tasks during startup

CHANGELOG.md

@@ -24,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Reduce and dependency tasks [#144](https://github.com/p2panda/aquadoggo/pull/144)
 - GraphQL endpoints for replication [#100](https://github.com/p2panda/aquadoggo/pull/100)
 - Inform materialization service about new operations [#161](https://github.com/p2panda/aquadoggo/pull/161)
+- Reschedule pending tasks on startup [#168](https://github.com/p2panda/aquadoggo/pull/168)
 
 ### Changed
 

aquadoggo/migrations/20220617115933_create-tasks.sql

@@ -0,0 +1,8 @@
+-- SPDX-License-Identifier: AGPL-3.0-or-later
+
+CREATE TABLE IF NOT EXISTS tasks (
+    name              TEXT      NOT NULL,
+    document_id       TEXT      NULL,
+    document_view_id  TEXT      NULL,
+    PRIMARY KEY (name, document_id, document_view_id)
+);

aquadoggo/src/db/errors.rs

@@ -1,9 +1,21 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-use p2panda_rs::{
-    document::{DocumentId, DocumentViewId},
-    schema::{system::SystemSchemaError, SchemaError, SchemaIdError},
-};
+use p2panda_rs::document::{DocumentId, DocumentViewId};
+use p2panda_rs::schema::system::SystemSchemaError;
+use p2panda_rs::schema::{SchemaError, SchemaIdError};
+
+/// `SQLStorage` errors.
+#[derive(thiserror::Error, Debug)]
+pub enum SqlStorageError {
+    #[error("SQL query failed: {0}")]
+    Transaction(String),
+
+    #[error("Insertion of row into table {0} did not show any effect")]
+    Insertion(String),
+
+    #[error("Deletion of row from table {0} did not show any effect")]
+    Deletion(String),
+}
 
 /// `DocumentStore` errors.
 #[derive(thiserror::Error, Debug)]

aquadoggo/src/db/models/mod.rs

@@ -4,7 +4,9 @@ pub mod document;
 mod entry;
 mod log;
 mod operation;
+mod task;
 
 pub use self::log::LogRow;
 pub use entry::EntryRow;
 pub use operation::{OperationFieldsJoinedRow, OperationRow};
+pub use task::TaskRow;

aquadoggo/src/db/models/task.rs

@@ -0,0 +1,20 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use serde::Serialize;
+use sqlx::FromRow;
+
+/// Representation of a row from the `tasks` table as stored in the database.
+///
+/// This table holds all "pending" tasks of the materialization service worker.
+#[derive(FromRow, Debug, Serialize, Clone, PartialEq)]
+#[serde(rename_all = "camelCase")]
+pub struct TaskRow {
+    /// Name of the task worker.
+    pub name: String,
+
+    /// `DocumentId` of the task input.
+    pub document_id: Option<String>,
+
+    /// `DocumentViewId` of the task input.
+    pub document_view_id: Option<String>,
+}

aquadoggo/src/db/provider.rs

@@ -1,14 +1,12 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 use async_trait::async_trait;
-use sqlx::query_scalar;
-
 use p2panda_rs::document::DocumentId;
 use p2panda_rs::hash::Hash;
 use p2panda_rs::storage_provider::traits::StorageProvider;
+use sqlx::query_scalar;
 
-use crate::db::stores::StorageEntry;
-use crate::db::stores::StorageLog;
+use crate::db::stores::{StorageEntry, StorageLog};
 use crate::db::Pool;
 use crate::errors::StorageProviderResult;
 use crate::graphql::client::{

aquadoggo/src/db/stores/mod.rs

@@ -5,6 +5,7 @@ mod entry;
 mod log;
 mod operation;
 mod schema;
+mod task;
 #[cfg(test)]
 pub mod test_utils;
 

aquadoggo/src/db/stores/task.rs

@@ -0,0 +1,210 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use anyhow::Result;
+use p2panda_rs::document::{DocumentId, DocumentViewId};
+use sqlx::{query, query_as};
+
+use crate::db::errors::SqlStorageError;
+use crate::db::models::TaskRow;
+use crate::db::provider::SqlStorage;
+use crate::materializer::{Task, TaskInput};
+
+/// Methods to interact with the `tasks` table in the database.
+impl SqlStorage {
+    /// Inserts a "pending" task into the database.
+    pub async fn insert_task(&self, task: &Task<TaskInput>) -> Result<(), SqlStorageError> {
+        // Convert task input to correct database types
+        let task_input = task.input();
+        let document_id = task_input.document_id.as_ref().map(|id| id.as_str());
+        let document_view_id = task_input
+            .document_view_id
+            .as_ref()
+            .map(|view_id| view_id.as_str());
+
+        // Check first if this task already exists, to avoid duplicate rows
+        let task_row = query_as::<_, TaskRow>(
+            "
+            SELECT
+                name,
+                document_id,
+                document_view_id
+            FROM
+                tasks
+            ",
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(|err| SqlStorageError::Transaction(err.to_string()))?;
+
+        // If yes, we are already done here
+        if task_row.is_some() {
+            return Ok(());
+        }
+
+        // Insert task into database
+        let result = query(
+            "
+            INSERT INTO
+                tasks (
+                    name,
+                    document_id,
+                    document_view_id
+                )
+            VALUES
+                ($1, $2, $3)
+            ",
+        )
+        .bind(task.worker_name())
+        .bind(document_id)
+        .bind(document_view_id)
+        .execute(&self.pool)
+        .await
+        .map_err(|err| SqlStorageError::Transaction(err.to_string()))?;
+
+        if result.rows_affected() != 1 {
+            Err(SqlStorageError::Insertion("tasks".into()))
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Removes a "pending" task from the database.
+    pub async fn remove_task(&self, task: &Task<TaskInput>) -> Result<(), SqlStorageError> {
+        // Convert task input to correct database types
+        let task_input = task.input();
+        let document_id = task_input.document_id.as_ref().map(|id| id.as_str());
+        let document_view_id = task_input
+            .document_view_id
+            .as_ref()
+            .map(|view_id| view_id.as_str());
+
+        // Remove task from database
+        let result = query(
+            "
+            DELETE FROM
+                tasks
+            WHERE
+                name = $1
+                AND document_id IS $2
+                AND document_view_id IS $3
+            ",
+        )
+        .bind(task.worker_name())
+        .bind(document_id)
+        .bind(document_view_id)
+        .execute(&self.pool)
+        .await
+        .map_err(|err| SqlStorageError::Transaction(err.to_string()))?;
+
+        if result.rows_affected() != 1 {
+            Err(SqlStorageError::Deletion("tasks".into()))
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Returns "pending" tasks of the materialization service worker.
+    pub async fn get_tasks(&self) -> Result<Vec<Task<TaskInput>>, SqlStorageError> {
+        let task_rows = query_as::<_, TaskRow>(
+            "
+            SELECT
+                name,
+                document_id,
+                document_view_id
+            FROM
+                tasks
+            ",
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(|err| SqlStorageError::Transaction(err.to_string()))?;
+
+        // Convert database rows into correct p2panda types
+        let mut tasks: Vec<Task<TaskInput>> = Vec::new();
+        for task in task_rows {
+            let document_id: Option<DocumentId> = task.document_id.map(|id| {
+                id.parse()
+                    .unwrap_or_else(|_| panic!("Invalid document id stored in database {}", id))
+            });
+
+            let document_view_id: Option<DocumentViewId> = task.document_view_id.map(|view_id| {
+                view_id.parse().unwrap_or_else(|_| {
+                    panic!("Invalid document view id stored in database: {}", view_id)
+                })
+            });
+
+            tasks.push(Task::new(
+                &task.name,
+                TaskInput::new(document_id, document_view_id),
+            ));
+        }
+
+        Ok(tasks)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use p2panda_rs::document::{DocumentId, DocumentViewId};
+    use p2panda_rs::test_utils::fixtures::{document_id, document_view_id};
+    use rstest::rstest;
+
+    use crate::db::stores::test_utils::{test_db, TestSqlStore};
+    use crate::materializer::{Task, TaskInput};
+
+    #[rstest]
+    #[tokio::test]
+    async fn insert_get_remove_tasks(
+        document_view_id: DocumentViewId,
+        #[from(test_db)]
+        #[future]
+        db: TestSqlStore,
+    ) {
+        let db = db.await;
+
+        // Prepare test data
+        let task = Task::new("reduce", TaskInput::new(None, Some(document_view_id)));
+
+        // Insert task
+        let result = db.store.insert_task(&task).await;
+        assert!(result.is_ok(), "{:?}", result);
+
+        // Check if task exists in database
+        let result = db.store.get_tasks().await;
+        assert_eq!(result.unwrap(), vec![task.clone()]);
+
+        // Remove task
+        let result = db.store.remove_task(&task).await;
+        assert!(result.is_ok(), "{:?}", result);
+
+        // Check if all tasks got removed
+        let result = db.store.get_tasks().await;
+        assert_eq!(result.unwrap(), vec![]);
+    }
+
+    #[rstest]
+    #[tokio::test]
+    async fn avoid_duplicates(
+        document_id: DocumentId,
+        #[from(test_db)]
+        #[future]
+        db: TestSqlStore,
+    ) {
+        let db = db.await;
+
+        // Prepare test data
+        let task = Task::new("reduce", TaskInput::new(Some(document_id), None));
+
+        // Insert task
+        let result = db.store.insert_task(&task).await;
+        assert!(result.is_ok(), "{:?}", result);
+
+        // Insert the same thing again, it should silently fail
+        let result = db.store.insert_task(&task).await;
+        assert!(result.is_ok(), "{:?}", result);
+
+        // Check for duplicates
+        let result = db.store.get_tasks().await;
+        assert_eq!(result.unwrap().len(), 1);
+    }
+}

aquadoggo/src/db/stores/test_utils.rs

@@ -168,8 +168,8 @@ pub async fn insert_entry_operation_and_view(
     (document_id, document_view_id)
 }
 
-/// Container for `SqlStore` with access to the document ids and key_pairs
-/// used in the pre-populated database for testing.
+/// Container for `SqlStore` with access to the document ids and key_pairs used in the
+/// pre-populated database for testing.
 pub struct TestSqlStore {
     pub store: SqlStorage,
     pub key_pairs: Vec<KeyPair>,

aquadoggo/src/db/utils.rs

@@ -328,7 +328,6 @@ pub fn parse_document_view_field_rows(
 
 #[cfg(test)]
 mod tests {
-
     use p2panda_rs::document::DocumentViewValue;
     use p2panda_rs::operation::{
         AsOperation, OperationId, OperationValue, PinnedRelation, PinnedRelationList, Relation,

aquadoggo/src/graphql/client/mutation.rs

@@ -65,9 +65,15 @@ impl ClientMutationRoot {
 
                 // Send new operation on service communication bus, this will arrive eventually at
                 // the materializer service
-                tx.send(ServiceMessage::NewOperation(
-                    verified_operation.operation_id().to_owned(),
-                ))?;
+                if tx
+                    .send(ServiceMessage::NewOperation(
+                        verified_operation.operation_id().to_owned(),
+                    ))
+                    .is_err()
+                {
+                    // Silently fail here as we don't mind if there are no subscribers. We have
+                    // tests in other places to check if messages arrive.
+                }
 
                 Ok(response)
             }

aquadoggo/src/materializer/input.rs

@@ -2,13 +2,23 @@
 
 use p2panda_rs::document::{DocumentId, DocumentViewId};
 
+/// Input of every task worker containing all information we need to process.
+///
+/// The workers are designed such that they EITHER await a `DocumentId` OR a `DocumentViewId`.
+/// Setting both values `None` or both values `Some` will be rejected.
 #[derive(Clone, Eq, PartialEq, Debug, Hash)]
 pub struct TaskInput {
+    /// Specifying a `DocumentId`, indicating that we're interested in processing the "latest"
+    /// state of that document.
     pub document_id: Option<DocumentId>,
+
+    /// Specifying a `DocumentViewId`, indicating that we're interested in processing the state of
+    /// that document view at this point.
     pub document_view_id: Option<DocumentViewId>,
 }
 
 impl TaskInput {
+    /// Returns a new instance of `TaskInput`.
     pub fn new(document_id: Option<DocumentId>, document_view_id: Option<DocumentViewId>) -> Self {
         Self {
             document_id,

aquadoggo/src/materializer/mod.rs

@@ -7,3 +7,4 @@ mod worker;
 
 pub use input::TaskInput;
 pub use service::materializer_service;
+pub use worker::Task;