Tests and fixes for oplog corruption bug (#962)

golemcloud · Sep 23, 2024 · 08eedaa · 08eedaa
1 parent a567b23
commit 08eedaa
Show file tree

Hide file tree

Showing 11 changed files with 353 additions and 32 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -94,7 +94,7 @@ futures-core = "0.3.29"
 futures-util = "0.3.29"
 git-version = "0.3.9"
 golem-wasm-ast = "1.0.0"
-golem-wasm-rpc = { version = "1.0.2", default-features = false, features = [
+golem-wasm-rpc = { version = "1.0.3", default-features = false, features = [
     "host",
 ] }
 http = "1.0.0" # keep in sync with wasmtime

diff --git a/golem-cli/Cargo.toml b/golem-cli/Cargo.toml
@@ -39,7 +39,7 @@ futures-util = { workspace = true }
 golem-examples = "1.0.5"
 golem-wasm-ast = { workspace = true }
 golem-wasm-rpc = { workspace = true }
-golem-wasm-rpc-stubgen = { version = "1.0.2", optional = true }
+golem-wasm-rpc-stubgen = { version = "1.0.3", optional = true }
 h2 = "0.3.24"
 http = { workspace = true }
 humansize = { workspace = true }

diff --git a/golem-worker-executor-base/src/services/oplog/mock.rs b/golem-worker-executor-base/src/services/oplog/mock.rs
@@ -49,7 +49,11 @@ impl OplogService for OplogServiceMock {
         unimplemented!()
     }
 
-    async fn open(&self, _owned_worker_id: &OwnedWorkerId) -> Arc<dyn Oplog + Send + Sync> {
+    async fn open(
+        &self,
+        _owned_worker_id: &OwnedWorkerId,
+        _last_oplog_index: OplogIndex,
+    ) -> Arc<dyn Oplog + Send + Sync> {
         unimplemented!()
     }
 

diff --git a/golem-worker-executor-base/src/services/oplog/mod.rs b/golem-worker-executor-base/src/services/oplog/mod.rs
@@ -71,8 +71,11 @@ pub trait OplogService: Debug {
         owned_worker_id: &OwnedWorkerId,
         initial_entry: OplogEntry,
     ) -> Arc<dyn Oplog + Send + Sync + 'static>;
-    async fn open(&self, owned_worker_id: &OwnedWorkerId)
-        -> Arc<dyn Oplog + Send + Sync + 'static>;
+    async fn open(
+        &self,
+        owned_worker_id: &OwnedWorkerId,
+        last_oplog_index: OplogIndex,
+    ) -> Arc<dyn Oplog + Send + Sync + 'static>;
 
     async fn get_last_index(&self, owned_worker_id: &OwnedWorkerId) -> OplogIndex;
 

diff --git a/golem-worker-executor-base/src/services/oplog/multilayer.rs b/golem-worker-executor-base/src/services/oplog/multilayer.rs
@@ -148,6 +148,7 @@ struct CreateOplogConstructor {
     initial_entry: Option<OplogEntry>,
     primary: Arc<dyn OplogService + Send + Sync>,
     service: MultiLayerOplogService,
+    last_oplog_index: OplogIndex,
 }
 
 impl CreateOplogConstructor {
@@ -156,12 +157,14 @@ impl CreateOplogConstructor {
         initial_entry: Option<OplogEntry>,
         primary: Arc<dyn OplogService + Send + Sync>,
         service: MultiLayerOplogService,
+        last_oplog_index: OplogIndex,
     ) -> Self {
         Self {
             owned_worker_id,
             initial_entry,
             primary,
             service,
+            last_oplog_index,
         }
     }
 }
@@ -177,7 +180,9 @@ impl OplogConstructor for CreateOplogConstructor {
                 .create(&self.owned_worker_id, initial_entry)
                 .await
         } else {
-            self.primary.open(&self.owned_worker_id).await
+            self.primary
+                .open(&self.owned_worker_id, self.last_oplog_index)
+                .await
         };
         Arc::new(MultiLayerOplog::new(self.owned_worker_id, primary, self.service, close).await)
     }
@@ -198,12 +203,18 @@ impl OplogService for MultiLayerOplogService {
                     Some(initial_entry),
                     self.primary.clone(),
                     self.clone(),
+                    OplogIndex::INITIAL,
                 ),
             )
             .await
     }
 
-    async fn open(&self, owned_worker_id: &OwnedWorkerId) -> Arc<dyn Oplog + Send + Sync> {
+    async fn open(
+        &self,
+        owned_worker_id: &OwnedWorkerId,
+        last_oplog_index: OplogIndex,
+    ) -> Arc<dyn Oplog + Send + Sync> {
+        debug!("MultiLayerOplogService::open {owned_worker_id}");
         self.oplogs
             .get_or_open(
                 &owned_worker_id.worker_id,
@@ -212,6 +223,7 @@ impl OplogService for MultiLayerOplogService {
                     None,
                     self.primary.clone(),
                     self.clone(),
+                    last_oplog_index,
                 ),
             )
             .await

diff --git a/golem-worker-executor-base/src/services/oplog/primary.rs b/golem-worker-executor-base/src/services/oplog/primary.rs
@@ -122,14 +122,17 @@ impl OplogService for PrimaryOplogService {
                 )
             });
 
-        self.open(owned_worker_id).await
+        self.open(owned_worker_id, OplogIndex::INITIAL).await
     }
 
-    async fn open(&self, owned_worker_id: &OwnedWorkerId) -> Arc<dyn Oplog + Send + Sync> {
+    async fn open(
+        &self,
+        owned_worker_id: &OwnedWorkerId,
+        last_oplog_index: OplogIndex,
+    ) -> Arc<dyn Oplog + Send + Sync> {
         record_oplog_call("open");
 
         let key = Self::oplog_key(&owned_worker_id.worker_id);
-        let last_oplog_index = self.get_last_index(owned_worker_id).await;
 
         self.oplogs
             .get_or_open(