implement 'delete_replica' replication job type

This commit adds a 'delete_replica' replication job type. It differs
from 'delete' type by only deleting the specific replica without marking
the repository deleted from the virtual storage completely.

While we could update 'delete' to be handled in a similar manner, this
would require us to do the work over multiple releases as old Praefect's
could delete repositories completely during an upgrade. For this reason,
let's implement a new job type on the side an remove the old 'delete' type
in future.

1 files changed, 59 insertions, 0 deletions

diff --git a/internal/praefect/replicator_pg_test.go b/internal/praefect/replicator_pg_test.go
index 55bf5a244..934e4a337 100644
--- a/internal/praefect/replicator_pg_test.go
+++ b/internal/praefect/replicator_pg_test.go
@@ -4,6 +4,7 @@ package praefect
 
 import (
 	"context"
+	"errors"
 	"net"
 	"path/filepath"
 	"testing"
@@ -66,3 +67,61 @@ func TestReplicatorInvalidSourceRepository(t *testing.T) {
 	require.NoError(t, err)
 	require.False(t, exists)
 }
+
+func TestReplicatorDestroy(t *testing.T) {
+	for _, tc := range []struct {
+		change datastore.ChangeType
+		exists bool
+		error  error
+	}{
+		{change: datastore.DeleteReplica, exists: true},
+		{change: datastore.DeleteRepo, exists: false},
+		{change: "invalid-type", exists: true, error: errors.New(`unknown change type: "invalid-type"`)},
+	} {
+		t.Run(string(tc.change), func(t *testing.T) {
+			db := getDB(t)
+
+			rs := datastore.NewPostgresRepositoryStore(db, nil)
+
+			ctx, cancel := testhelper.Context()
+			defer cancel()
+
+			require.NoError(t, rs.SetGeneration(ctx, "virtual-storage-1", "relative-path-1", "storage-1", 0))
+			require.NoError(t, rs.SetGeneration(ctx, "virtual-storage-1", "relative-path-1", "storage-2", 0))
+
+			ln, err := net.Listen("tcp", "localhost:0")
+			require.NoError(t, err)
+
+			srv := grpc.NewServer(grpc.UnknownServiceHandler(func(srv interface{}, stream grpc.ServerStream) error {
+				return stream.SendMsg(&gitalypb.RemoveRepositoryResponse{})
+			}))
+
+			go srv.Serve(ln)
+			defer srv.Stop()
+
+			clientConn, err := grpc.Dial(ln.Addr().String(), grpc.WithInsecure())
+			require.NoError(t, err)
+			defer clientConn.Close()
+
+			require.Equal(t, tc.error, defaultReplicator{
+				rs:  rs,
+				log: testhelper.DiscardTestLogger(t),
+			}.Destroy(
+				ctx,
+				datastore.ReplicationEvent{
+					Job: datastore.ReplicationJob{
+						Change:            tc.change,
+						VirtualStorage:    "virtual-storage-1",
+						RelativePath:      "relative-path-1",
+						TargetNodeStorage: "storage-1",
+					},
+				},
+				clientConn,
+			))
+
+			exists, err := rs.RepositoryExists(ctx, "virtual-storage-1", "relative-path-1")
+			require.NoError(t, err)
+			require.Equal(t, tc.exists, exists)
+		})
+	}
+}