Recover a diverged repository from a source replica

.coderabbit.yaml

@@ -5,3 +5,7 @@ reviews:
     enabled: true
     drafts: true
     ignore_title_keywords: ["WIP"]
+    # A feature branch collects its pull requests on an integration branch before it is merged into
+    # main, so review those pull requests too, not only the ones that target the default branch.
+    base_branches:
+      - ".*"

server/src/main/java/com/linecorp/centraldogma/server/CentralDogma.java

@@ -171,6 +171,7 @@
 import com.linecorp.centraldogma.server.internal.mirror.DefaultMirroringServicePlugin;
 import com.linecorp.centraldogma.server.internal.mirror.MirrorAccessControl;
 import com.linecorp.centraldogma.server.internal.mirror.MirrorRunner;
+import com.linecorp.centraldogma.server.internal.replication.RecoveryPayloadBuilder;
 import com.linecorp.centraldogma.server.internal.replication.ZooKeeperCommandExecutor;
 import com.linecorp.centraldogma.server.internal.storage.project.DefaultProjectManager;
 import com.linecorp.centraldogma.server.internal.storage.project.ProjectApiManager;
@@ -963,7 +964,7 @@ private CommandExecutor newZooKeeperCommandExecutor(
                                               sessionManager, encryptionStorageManager,
                         /* onTakeLeadership */ null, /* onReleaseLeadership */ null,
                         /* onTakeZoneLeadership */ null, /* onReleaseZoneLeadership */ null),
-                meterRegistry, zone,
+                meterRegistry, zone, new RecoveryPayloadBuilder(pm),
                 onTakeLeadership, onReleaseLeadership,
                 onTakeZoneLeadership, onReleaseZoneLeadership);
     }
@@ -1028,7 +1029,8 @@ private void configureHttpApi(ServerBuilder sb,
                 .annotatedService(new ServerStatusService(executor, statusManager, repoStatusManager))
                 .annotatedService(new ProjectServiceV1(projectApiManager, executor, repoStatusManager))
                 .annotatedService(new RepositoryServiceV1(executor, mds, encryptionStorageManager,
-                                                          repoStatusManager))
+                                                          repoStatusManager,
+                                                          new RecoveryPayloadBuilder(pm)))
                 .annotatedService(new CredentialServiceV1(projectApiManager, executor))
                 .annotatedService(new VariableServiceV1(pm, executor));
         if (LOGBACK_ENABLED) {

server/src/main/java/com/linecorp/centraldogma/server/command/Command.java

@@ -68,6 +68,8 @@
         @Type(value = UpdateProjectStatusCommand.class, name = "UPDATE_PROJECT_STATUS"),
         @Type(value = UpdateRepositoryStatusCommand.class, name = "UPDATE_REPOSITORY_STATUS"),
         @Type(value = ForcePushCommand.class, name = "FORCE_PUSH_COMMAND"),
+        @Type(value = RecoverRepositoryCommand.class, name = "RECOVER_REPOSITORY"),
+        @Type(value = RecoverRepositoryRequestCommand.class, name = "RECOVER_REPOSITORY_REQUEST"),
 })
 public interface Command<T> {
 
@@ -526,6 +528,38 @@ static <T> Command<T> forcePush(Command<T> delegate) {
         return new ForcePushCommand<>(delegate);
     }
 
+    /**
+     * Returns a new {@link Command} which recovers a diverged repository from a source replica by resetting
+     * to {@code resetToRevision} and replaying {@code commits} up to {@code headRevision}. See
+     * {@link RecoverRepositoryCommand}.
+     */
+    static Command<Revision> recoverRepository(Author author, String projectName, String repositoryName,
+                                               int sourceServerId, Revision resetToRevision,
+                                               Revision headRevision, Iterable<ReplayCommit> commits) {
+        requireNonNull(author, "author");
+        requireNonNull(projectName, "projectName");
+        requireNonNull(repositoryName, "repositoryName");
+        requireNonNull(resetToRevision, "resetToRevision");
+        requireNonNull(headRevision, "headRevision");
+        requireNonNull(commits, "commits");
+        return new RecoverRepositoryCommand(null, author, projectName, repositoryName, sourceServerId,
+                                            resetToRevision, headRevision, commits);
+    }
+
+    /**
+     * Returns a new {@link Command} which asks the source replica to originate a recovery. See
+     * {@link RecoverRepositoryRequestCommand}.
+     */
+    static Command<Void> recoverRepositoryRequest(Author author, String projectName, String repositoryName,
+                                                  int sourceServerId, Revision fromRevision) {
+        requireNonNull(author, "author");
+        requireNonNull(projectName, "projectName");
+        requireNonNull(repositoryName, "repositoryName");
+        requireNonNull(fromRevision, "fromRevision");
+        return new RecoverRepositoryRequestCommand(null, author, projectName, repositoryName,
+                                                   sourceServerId, fromRevision);
+    }
+
     /**
      * Returns the {@link CommandType} of the command.
      */

server/src/main/java/com/linecorp/centraldogma/server/command/CommandType.java

@@ -49,6 +49,8 @@ public enum CommandType {
     UPDATE_SERVER_STATUS(Void.class),
     UPDATE_PROJECT_STATUS(Void.class),
     UPDATE_REPOSITORY_STATUS(Void.class),
+    RECOVER_REPOSITORY(Revision.class),
+    RECOVER_REPOSITORY_REQUEST(Void.class),
     // The result type of FORCE_PUSH is Object because it can be any type.
     FORCE_PUSH(Object.class);
 

server/src/main/java/com/linecorp/centraldogma/server/command/RecoverRepositoryCommand.java

@@ -0,0 +1,137 @@
+/*
+ * Copyright 2026 LY Corporation
+ *
+ * LY Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.linecorp.centraldogma.server.command;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.List;
+import java.util.Objects;
+
+import org.jspecify.annotations.Nullable;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.MoreObjects.ToStringHelper;
+import com.google.common.collect.ImmutableList;
+
+import com.linecorp.centraldogma.common.Author;
+import com.linecorp.centraldogma.common.Revision;
+
+/**
+ * A {@link Command} which recovers a diverged repository from a source replica. It is originated by the
+ * source replica (the single source of truth) and applied identically on every replica, itself included: a
+ * replica already converged with {@link #commits()} is left untouched, and every other one resets its git
+ * repository and commit-id database to {@link #resetToRevision()} and replays {@link #commits()} up to
+ * {@link #headRevision()}. Because the replayed commits carry the original author, timestamp and
+ * self-contained changes, a replay reproduces the source's commit ids; each one is verified against
+ * {@link ReplayCommit#expectedCommitId()}, and a mismatch aborts the recovery and rolls the replica back.
+ *
+ * <p>The convergence check is by content, not by replica: the source is normally the replica that is
+ * already converged, but a commit that lands on it between the payload build and the apply (a force push,
+ * which read-only does not block) makes the source replay over itself too, discarding that commit.
+ *
+ * <p>This is a {@link RepositoryCommand} so that it is scoped to a single repository (lock scope and
+ * read-only failure blast radius) and is not rejected while the repository/project is read-only.
+ */
+public final class RecoverRepositoryCommand extends RepositoryCommand<Revision> {
+
+    private final int sourceServerId;
+    private final Revision resetToRevision;
+    private final Revision headRevision;
+    private final List<ReplayCommit> commits;
+
+    @JsonCreator
+    RecoverRepositoryCommand(@JsonProperty("timestamp") @Nullable Long timestamp,
+                             @JsonProperty("author") @Nullable Author author,
+                             @JsonProperty("projectName") String projectName,
+                             @JsonProperty("repositoryName") String repositoryName,
+                             @JsonProperty("sourceServerId") int sourceServerId,
+                             @JsonProperty("resetToRevision") Revision resetToRevision,
+                             @JsonProperty("headRevision") Revision headRevision,
+                             @JsonProperty("commits") Iterable<ReplayCommit> commits) {
+        super(CommandType.RECOVER_REPOSITORY, timestamp, author, projectName, repositoryName);
+        this.sourceServerId = sourceServerId;
+        this.resetToRevision = requireNonNull(resetToRevision, "resetToRevision");
+        this.headRevision = requireNonNull(headRevision, "headRevision");
+        this.commits = ImmutableList.copyOf(requireNonNull(commits, "commits"));
+    }
+
+    /**
+     * Returns the ZooKeeper server ID of the source replica whose repository the {@link #commits()} were
+     * taken from. It records where a recovery came from; it is not consulted when the command is applied,
+     * which decides by content (see the class javadoc).
+     */
+    @JsonProperty
+    public int sourceServerId() {
+        return sourceServerId;
+    }
+
+    /**
+     * Returns the {@link Revision} to which a replica resets its repository before replaying
+     * {@link #commits()}.
+     */
+    @JsonProperty
+    public Revision resetToRevision() {
+        return resetToRevision;
+    }
+
+    /**
+     * Returns the head {@link Revision} of the source repository, which is also the result of this command.
+     */
+    @JsonProperty
+    public Revision headRevision() {
+        return headRevision;
+    }
+
+    /**
+     * Returns the ordered {@link ReplayCommit}s to replay after resetting to {@link #resetToRevision()}.
+     */
+    @JsonProperty
+    public List<ReplayCommit> commits() {
+        return commits;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (!(obj instanceof RecoverRepositoryCommand)) {
+            return false;
+        }
+        final RecoverRepositoryCommand that = (RecoverRepositoryCommand) obj;
+        return super.equals(that) &&
+               sourceServerId == that.sourceServerId &&
+               resetToRevision.equals(that.resetToRevision) &&
+               headRevision.equals(that.headRevision) &&
+               commits.equals(that.commits);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(sourceServerId, resetToRevision, headRevision, commits) * 31 + super.hashCode();
+    }
+
+    @Override
+    ToStringHelper toStringHelper() {
+        return super.toStringHelper()
+                    .add("sourceServerId", sourceServerId)
+                    .add("resetToRevision", resetToRevision)
+                    .add("headRevision", headRevision)
+                    .add("commits", commits.size());
+    }
+}

server/src/main/java/com/linecorp/centraldogma/server/command/RecoverRepositoryRequestCommand.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2026 LY Corporation
+ *
+ * LY Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.linecorp.centraldogma.server.command;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.Objects;
+
+import org.jspecify.annotations.Nullable;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.MoreObjects.ToStringHelper;
+
+import com.linecorp.centraldogma.common.Author;
+import com.linecorp.centraldogma.common.Revision;
+
+/**
+ * A {@link Command} which asks the source replica to originate a {@link RecoverRepositoryCommand}. It is
+ * originated by a non-source replica that received the recovery request (e.g. behind a load balancer) and is
+ * applied as a no-op on every replica; the source replica reacts to it (off the replication-log replay
+ * thread) by building and originating the actual {@link RecoverRepositoryCommand}.
+ */
+public final class RecoverRepositoryRequestCommand extends RepositoryCommand<Void> {
+
+    private final int sourceServerId;
+    private final Revision fromRevision;
+
+    @JsonCreator
+    RecoverRepositoryRequestCommand(@JsonProperty("timestamp") @Nullable Long timestamp,
+                                    @JsonProperty("author") @Nullable Author author,
+                                    @JsonProperty("projectName") String projectName,
+                                    @JsonProperty("repositoryName") String repositoryName,
+                                    @JsonProperty("sourceServerId") int sourceServerId,
+                                    @JsonProperty("fromRevision") Revision fromRevision) {
+        super(CommandType.RECOVER_REPOSITORY_REQUEST, timestamp, author, projectName, repositoryName);
+        this.sourceServerId = sourceServerId;
+        this.fromRevision = requireNonNull(fromRevision, "fromRevision");
+    }
+
+    /**
+     * Returns the ZooKeeper server ID of the source replica that should originate the recovery.
+     */
+    @JsonProperty
+    public int sourceServerId() {
+        return sourceServerId;
+    }
+
+    /**
+     * Returns the first {@link Revision} to replay. Recovery replays {@code fromRevision..sourceHead}.
+     */
+    @JsonProperty
+    public Revision fromRevision() {
+        return fromRevision;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (!(obj instanceof RecoverRepositoryRequestCommand)) {
+            return false;
+        }
+        final RecoverRepositoryRequestCommand that = (RecoverRepositoryRequestCommand) obj;
+        return super.equals(that) &&
+               sourceServerId == that.sourceServerId &&
+               fromRevision.equals(that.fromRevision);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(sourceServerId, fromRevision) * 31 + super.hashCode();
+    }
+
+    @Override
+    ToStringHelper toStringHelper() {
+        return super.toStringHelper()
+                    .add("sourceServerId", sourceServerId)
+                    .add("fromRevision", fromRevision);
+    }
+}