[DilwoarH]

** 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

flask shell

import json;
from app.dao.notifications_wal_changes_dao import dao_process_notifications_replication_slot_changes;
result = dao_process_notifications_replication_slot_changes();
print(json.dumps(result, indent=2))

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

[DilwoarH]

requires #4936 to be merged first

[DilwoarH]

@risicle

The following changes have been made:

• reorder batch_last_lsn check so that it happens before any return
• removed additional checks before acquiring lock
• removed additional advancing slot logic and let main flow handle processing
• removed celery schedule so that it doesn't start processing WAL entries (we will use an alternative method)

Rob is reviewing

[risicle]

😕 what happens if it's not?

If this is just to please mypy, we could do

assert isinstance(payload, str) # type narrowing

which is something we've done elsewhere.

[DilwoarH]

updated with new logic

[risicle]

Is this function needed anymore?

[DilwoarH]

done

[DilwoarH]

@risicle

You can test this locally using:

flask shell

import json;
from app.dao.notifications_wal_changes_dao import dao_process_notifications_replication_slot_changes;
result = dao_process_notifications_replication_slot_changes();
print(json.dumps(result, indent=2));