Replication job acknowledge removes 'completed' and 'dead' events.

Migration script to remove all redundant rows from replication_queue.
It is safe to re-run migration if it fails. It is designed to remove
rows in batches by 100000 each iteration. Each removal happens in
separate transaction. The script removes only 'dead' and 'completed'
replication events.

Closes: https://gitlab.com/gitlab-org/gitaly/-/issues/2941

2 files changed, 44 insertions, 0 deletions

diff --git a/changelogs/unreleased/ps-ack-does-delete.yml b/changelogs/unreleased/ps-ack-does-delete.yml
new file mode 100644
index 000000000..423cd092a
--- /dev/null
+++ b/changelogs/unreleased/ps-ack-does-delete.yml
@@ -0,0 +1,5 @@
+---
+title: Replication job acknowledge removes 'completed' and 'dead' events.
+merge_request: 2457
+author:
+type: added
diff --git a/internal/praefect/datastore/migrations/20200810055650_replication_queue_cleanup.go b/internal/praefect/datastore/migrations/20200810055650_replication_queue_cleanup.go
new file mode 100644
index 000000000..4dcf2fed4
--- /dev/null
+++ b/internal/praefect/datastore/migrations/20200810055650_replication_queue_cleanup.go
@@ -0,0 +1,39 @@
+package migrations
+
+import migrate "github.com/rubenv/sql-migrate"
+
+func init() {
+	m := &migrate.Migration{
+		Id: "20200810055650_replication_queue_cleanup",
+		Up: []string{
+			`CREATE TEMP TABLE ids(id BIGINT)`,
+			`
+			-- +migrate StatementBegin
+			DO $$DECLARE
+			    found_amount BIGINT;
+			BEGIN
+			    LOOP
+				TRUNCATE TABLE ids;
+				INSERT INTO ids
+				SELECT id
+				FROM replication_queue
+				WHERE state = ANY (ARRAY['dead'::REPLICATION_JOB_STATE, 'completed'::REPLICATION_JOB_STATE])
+				LIMIT 100000;
+				SELECT COUNT(*) INTO found_amount FROM ids;
+				IF found_amount > 0 THEN
+				    DELETE FROM replication_queue WHERE id IN (SELECT id FROM ids);
+				    COMMIT;
+				ELSE
+				    RETURN;
+				END IF;
+			    END LOOP;
+			END$$;
+			-- +migrate StatementEnd`,
+			`DROP TABLE ids`,
+		},
+		Down:                 []string{},
+		DisableTransactionUp: true,
+	}
+
+	allMigrations = append(allMigrations, m)
+}