[server] Fix duplicate log reconciliation messages #1438 (#1441)

* Update forward.rs * Update agdb_server.yaml * fix conditions
agnesoft · Jan 2, 2025 · db42939 · db42939
1 parent 0fe7a8a
commit db42939
Show file tree

Hide file tree

Showing 4 changed files with 39 additions and 28 deletions.
diff --git a/.github/workflows/agdb_server.yaml b/.github/workflows/agdb_server.yaml
@@ -36,7 +36,7 @@ jobs:
       - uses: actions-rust-lang/setup-rust-toolchain@v1
       - uses: taiki-e/install-action@cargo-llvm-cov
       - run: rustup component add llvm-tools-preview
-      - run: cargo llvm-cov --package agdb_server --all-features --ignore-filename-regex "agdb(.|..)src|agdb_derive|agdb_api|api.rs" --show-missing-lines
+      - run: cargo llvm-cov --package agdb_server --all-features --ignore-filename-regex "agdb(.|..)src|agdb_derive|agdb_api|api.rs" --fail-uncovered-functions 17 --show-missing-lines
 
   agdb_server_test:
     runs-on: ubuntu-latest

diff --git a/agdb_server/src/forward.rs b/agdb_server/src/forward.rs
@@ -67,7 +67,7 @@ pub(crate) async fn forward_to_leader(
         } else {
             return Response::builder()
                 .status(StatusCode::SERVICE_UNAVAILABLE)
-                .body("Cluster is not ready yet".into())
+                .body("cluster is not ready yet".into())
                 .expect("cluster not ready yet response");
         }
     }

diff --git a/agdb_server/src/raft.rs b/agdb_server/src/raft.rs
@@ -340,12 +340,13 @@ impl<T: Clone, N, S: Storage<T, N>> Cluster<T, N, S> {
         );
 
         for log in logs {
-            self.validate_log_append(request, log)?;
-            self.append_storage(log)
-                .await
-                .map_err(|e| self.commit_error(request, e.description))?;
+            if self.validate_log_append(request, log)? {
+                self.append_storage(log)
+                    .await
+                    .map_err(|e| self.commit_error(request, e.description))?;
+            }
 
-            if log.index <= request.log_commit {
+            if log.index <= request.log_commit && self.local().log_commit < log.index {
                 self.commit_storage(log.index)
                     .await
                     .map_err(|e| self.commit_error(request, e.description))?;
@@ -559,28 +560,38 @@ impl<T: Clone, N, S: Storage<T, N>> Cluster<T, N, S> {
         Ok(())
     }
 
-    fn validate_log_append(&self, request: &Request<T>, log: &Log<T>) -> Result<(), Response> {
-        if self.local().log_commit >= log.index || log.index > (self.local().log_index + 1) {
-            return Err(Response {
-                target: request.index,
-                result: ResponseType::LogMismatch(LogMismatch {
-                    index: MismatchedValues {
-                        local: Some(self.local().log_index),
-                        requested: Some(log.index),
-                    },
-                    term: MismatchedValues {
-                        local: Some(self.local().log_term),
-                        requested: Some(log.term),
-                    },
-                    commit: MismatchedValues {
-                        local: Some(self.local().log_commit),
-                        requested: Some(log.index),
-                    },
-                }),
-            });
+    fn validate_log_append(&self, request: &Request<T>, log: &Log<T>) -> Result<bool, Response> {
+        if self.local().log_term == log.term {
+            if self.local().log_index >= log.index {
+                return Ok(false);
+            } else if self.local().log_commit < log.index && self.local().log_index + 1 == log.index
+            {
+                return Ok(true);
+            }
+        } else if self.local().log_term < log.term
+            && self.local().log_commit < log.index
+            && self.local().log_index + 1 >= log.index
+        {
+            return Ok(true);
         }
 
-        Ok(())
+        Err(Response {
+            target: request.index,
+            result: ResponseType::LogMismatch(LogMismatch {
+                index: MismatchedValues {
+                    local: Some(self.local().log_index),
+                    requested: Some(log.index),
+                },
+                term: MismatchedValues {
+                    local: Some(self.local().log_term),
+                    requested: Some(log.term),
+                },
+                commit: MismatchedValues {
+                    local: Some(self.local().log_commit),
+                    requested: Some(log.index),
+                },
+            }),
+        })
     }
 
     fn validate_log_for_vote(&self, request: &Request<T>) -> Result<(), Response> {

diff --git a/agdb_server/src/server_db.rs b/agdb_server/src/server_db.rs
@@ -509,7 +509,7 @@ impl ServerDb {
             let logs: Vec<DbId> = t
                 .exec(
                     QueryBuilder::select()
-                        .values(["index", "term"])
+                        .values("index")
                         .search()
                         .index(COMMITTED)
                         .value(false)