Process replication slot changes by DilwoarH · Pull Request #4930 · alphagov/notifications-api

DilwoarH · 2026-08-05T12:17:25Z

** DO NOT MERGE UNTIL #4946 IS MERGED IN FIRST**

What

The application code for processing slot changes for notifications table

Why

This is needed to process WAL logs in the database

Test results

Ran a load test on dev-c - this is the result after 2400 emails.

SELECT
	(SELECT count(*) FROM pg_logical_slot_peek_changes('notify_dashboard_replication_slot', NULL, NULL)) AS still_processing,
    (SELECT SUM(notification_count) FROM ft_service_stats) AS ft_stats_count,
    COUNT(*) AS notif_count,
    (SELECT SUM(notification_count) FROM ft_service_stats where notification_status = 'created') AS ft_created_count,
    COUNT(CASE WHEN notification_status = 'created' THEN 1 END) AS notif_created_count,
    (SELECT SUM(notification_count) FROM ft_service_stats where notification_status = 'sending') AS ft_sending_count,
    COUNT(CASE WHEN notification_status = 'sending' THEN 1 END) AS notif_sending_count,
    (SELECT SUM(notification_count) FROM ft_service_stats where notification_status = 'delivered') AS ft_delivered_count,
    COUNT(CASE WHEN notification_status = 'delivered' THEN 1 END) AS notif_delivered_count,
    (SELECT SUM(notification_count) FROM ft_service_stats where notification_status = 'temporary-failure') AS ft_temp_fail_count,
    COUNT(CASE WHEN notification_status = 'temporary-failure' THEN 1 END) AS notif_temp_fail_count,
    (SELECT SUM(notification_count) FROM ft_service_stats where notification_status = 'permanent-failure') AS ft_perm_fail_count,
    COUNT(CASE WHEN notification_status = 'permanent-failure' THEN 1 END) AS notif_perm_fail_count
FROM
    public.notifications;

still_processing	ft_stats_count	notif_count	ft_created_count	notif_created_count	ft_sending_count	notif_sending_count	ft_delivered_count	notif_delivered_count	ft_temp_fail_count	notif_temp_fail_count	ft_perm_fail_count	notif_perm_fail_count
0	2400	2400	0	0	0	0	1206	1206	234	234	960	960

How it works

605101533-57f970e2-60d7-462a-9fbd-b24d553d6663

DilwoarH · 2026-08-18T00:19:38Z

requires #4936 to be merged first

spatel033 · 2026-08-24T14:50:27Z

+    if key_type == "test":
+        return None
+
+    if require_status_from_primary_row and not notification_status:


Do we actually need this condition check? Updated notifications always have a previous notification_status from the database, this condition seems redundant and a bit confusing.

This is needed, however, it wasn't working as expected. I have fixed it and added some comments to support it

DilwoarH · 2026-09-17T10:21:49Z

+
+
+@notify_celery.task(name="process-notifications-replication-slot-changes")
+def process_replication_slot_changes():


to do - rename this to match task name

whpearson · 2026-09-21T13:05:04Z

+            .filter(*filters)
+            .update(
+                {
+                    "notification_count": func.greatest(FactServiceStats.notification_count + change_count, 0),


Should we error if change_count is bigger than notification_count as this shouldn't happen?

Note: We may need to test what happens if the row doesn't exist at all

DilwoarH · 2026-09-21T13:29:47Z

+                  AND pid = pg_backend_pid()
+                  AND granted
+                  AND classid = CAST((CAST(:lock_id AS bigint) >> 32) AS integer)
+                  AND objid   = CAST((CAST(:lock_id AS bigint) & 4294967295) AS integer)


4294967295

^^ document what this means

also the 32

DilwoarH · 2026-09-21T13:56:42Z

+
+
+def _extract_row_data(change: dict[str, Any]) -> RowData:
+    if "columnnames" in change and "columnvalues" in change:


This is only used in format v1 therefore we can remove now

DilwoarH · 2026-09-21T14:00:08Z

+    if "identity" in change and change.get("action") in {"D", "U", "I"}:
+        return _extract_name_value_rows(change["identity"])


not needed for insert

DilwoarH · 2026-09-21T14:00:25Z

+    oldkeys = change.get("oldkeys") or {}
+    if "keynames" in oldkeys and "keyvalues" in oldkeys:
+        return _zip_values(oldkeys["keynames"], oldkeys["keyvalues"])
+
+    if "keys" in oldkeys:
+        return _extract_name_value_rows(oldkeys["keys"])


DilwoarH · 2026-09-21T14:06:40Z

+        if change.get("lsn"):
+            last_lsn = change["lsn"]
+        elif change.get("nextlsn"):
+            last_lsn = change["nextlsn"]


This may be able to refactor out

DilwoarH · 2026-09-22T12:21:25Z

+        if table_name != REPLICATION_SLOT_TABLE_NAME.split(".")[-1]:
+            ignored_changes += 1
+            continue


This shouldn't ever happen - so can be removed

DilwoarH · 2026-09-22T12:23:49Z

+    # If the key_type is "test", we ignore this change and return None to indicate
+    # that it should not be processed further.
+    if key_type == "test":
+        return None


Potentially could be refactored to happen at an earlier stage

DilwoarH · 2026-09-22T12:27:52Z

+            if old_dimensions:
+                counter[old_dimensions] -= 1
+                updated = True


If no old dimension - we should continue because we don't want to increase the same status again

This will be replaced with the actual task in the following commits

the class is not processing all replication slot changes - just notifications

These are not needed

…eplication slot changes

Also include bst_date in the dimensions

…ed tests

…t stats updates and slot advancement in a single transaction

…eplication slot accordingly

… advancement and update related tests

This is to avoid unnessary calls

10 seconds sometimes feels delayed for users

DilwoarH mentioned this pull request Aug 5, 2026

[DO NOT MERGE] Add Service Stats (Wal2JSON implementation) #4873

Closed

DilwoarH force-pushed the process-replication-slot-changes branch 6 times, most recently from 6bddcf6 to 2d6790a Compare August 11, 2026 09:53

DilwoarH force-pushed the process-replication-slot-changes branch 2 times, most recently from da47acd to 1f988ae Compare August 18, 2026 00:15

DilwoarH changed the title ~~[WIP] Process replication slot changes~~ Process replication slot changes Aug 18, 2026

DilwoarH force-pushed the process-replication-slot-changes branch from 1f988ae to 1d032b9 Compare August 18, 2026 00:29

spatel033 reviewed Aug 19, 2026

View reviewed changes

Comment thread app/dao/fact_service_stats_dao.py Outdated

spatel033 reviewed Aug 24, 2026

View reviewed changes

Comment thread app/dao/notifications_wal_changes_dao.py

spatel033 reviewed Aug 24, 2026

View reviewed changes

DilwoarH mentioned this pull request Sep 2, 2026

Add migration to create logical replication slot #4946

Open

DilwoarH force-pushed the process-replication-slot-changes branch 3 times, most recently from 09efa7c to 29d32f3 Compare September 8, 2026 12:09

DilwoarH commented Sep 17, 2026

View reviewed changes

whpearson reviewed Sep 21, 2026

View reviewed changes

DilwoarH commented Sep 21, 2026

View reviewed changes

DilwoarH commented Sep 22, 2026

View reviewed changes

add a placeholder schedule to process replication slot changes

a6c2bc1

This will be replaced with the actual task in the following commits

DilwoarH and others added 29 commits September 25, 2026 10:42

move lock acquisition inside of the try catch

77ced53

get replication changes

26224ad

renamed file to match non generic purpose

f8d6c2c

the class is not processing all replication slot changes - just notifications

Return structure for no replication slot changes found

64d5ac0

Implement processing and logging for replication slot changes

df1897b

Remove unused NIL_UUID constant from notifications_wal_changes_dao.py

ceebc23

Refactor FullDimensions tuple by removing job_id and key_type

a4b31b8

These are not needed

Add service stats change counts aggregation and improve logging for r…

2ce4af8

…eplication slot changes

Add FactServiceStats model

9a18fd2

Add DAO to update ft_service_stats

868d7cb

Enhance service stats processing in replication slot changes

94c2c0a

Also include bst_date in the dimensions

Advance replication slot after insert

5b8d391

Remove debug logging

35e67df

Add celerybeat-schedule wildcard to .gitignore

0d05721

Refactor code to ensure it meets ruff standards

0c073f2

Add unit tests for fact_service_stats_dao function

ba19ee8

Rename replication slot processing task and add tests

5249688

Add unit tests for notifications_wal_changes_dao functions

bc46696

Support format version 2 in WAL2JSON payload parsing and update relat…

ba5c4d6

…ed tests

log start and end time metrics

3a4fe0f

Fix bug where last lsn doesn't exist

921c4f3

Refactor transaction handling in replication slot processing to commi…

a73ee64

…t stats updates and slot advancement in a single transaction

Handle non-indexable result rows in replication changes and advance r…

11d74b0

…eplication slot accordingly

Refactor replication slot processing to use SQL-reported LSN for slot…

9314e36

… advancement and update related tests

Change ordering of variable to ensure consistency

72815a9

Add comment to provide additional context

f1c89c7

Move filters into else statement

c0e4218

This is to avoid unnessary calls

add test with simulated traffic

f8897ea

change to 5 seconds

a1c7a1f

10 seconds sometimes feels delayed for users

DilwoarH force-pushed the process-replication-slot-changes branch from 29d32f3 to a1c7a1f Compare September 25, 2026 09:43



		@notify_celery.task(name="process-notifications-replication-slot-changes")
		def process_replication_slot_changes():



		def _extract_row_data(change: dict[str, Any]) -> RowData:
		if "columnnames" in change and "columnvalues" in change:

		if "identity" in change and change.get("action") in {"D", "U", "I"}:
		return _extract_name_value_rows(change["identity"])

Conversation

DilwoarH commented Aug 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What

Why

Test results

How it works

Uh oh!

DilwoarH commented Aug 18, 2026

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

DilwoarH commented Aug 5, 2026 •

edited

Loading