fix(sync): let the server own updated_at on insert for cursor-pulled tables
This commit is contained in:
Yun Chan
parent
2251fe2da1
commit
f517eedffc
2 changed files with 265 additions and 0 deletions
|
|
@ -0,0 +1,68 @@
|
||||||
|
-- Server-owned sync timestamps.
|
||||||
|
--
|
||||||
|
-- The desktop mirror (SyncEngine.pullEntity) and the audio sync read every
|
||||||
|
-- synced table with a (updated_at, id) keyset cursor and rewind it by only a
|
||||||
|
-- minute. That cursor is only sound when updated_at is a server clock value
|
||||||
|
-- for every write. moddatetime already owns it on UPDATE, but an INSERT kept
|
||||||
|
-- whatever updated_at the client sent:
|
||||||
|
--
|
||||||
|
-- a) a row recorded offline at T0 and pushed a day later landed behind every
|
||||||
|
-- other desktop's cursor and was never pulled there;
|
||||||
|
-- b) a desktop whose clock ran ahead inserted rows in the future, pushed the
|
||||||
|
-- other desktops' cursors ahead with them, and every server-stamped edit in
|
||||||
|
-- that window was then skipped for good;
|
||||||
|
-- c) restore_account_portability re-inserted archived updated_at values, so a
|
||||||
|
-- desktop that already applied the delete tombstones never got the rows
|
||||||
|
-- back.
|
||||||
|
--
|
||||||
|
-- From here on the server stamps updated_at on INSERT as well. user_templates
|
||||||
|
-- had no UPDATE stamp at all (only its RPC set it), so it gets one too. The
|
||||||
|
-- value is now(), the same clock moddatetime uses, so INSERT and UPDATE share
|
||||||
|
-- one cursor timeline and the existing overlap window still covers late
|
||||||
|
-- commits.
|
||||||
|
|
||||||
|
BEGIN;
|
||||||
|
|
||||||
|
CREATE OR REPLACE FUNCTION public.stamp_sync_timestamp_v1()
|
||||||
|
RETURNS trigger
|
||||||
|
LANGUAGE plpgsql
|
||||||
|
SET search_path = ''
|
||||||
|
AS $$
|
||||||
|
BEGIN
|
||||||
|
NEW.updated_at := pg_catalog.now();
|
||||||
|
RETURN NEW;
|
||||||
|
END;
|
||||||
|
$$;
|
||||||
|
|
||||||
|
REVOKE ALL ON FUNCTION public.stamp_sync_timestamp_v1() FROM PUBLIC, anon, authenticated;
|
||||||
|
|
||||||
|
DO $$
|
||||||
|
DECLARE
|
||||||
|
t text;
|
||||||
|
BEGIN
|
||||||
|
FOREACH t IN ARRAY ARRAY[
|
||||||
|
'history',
|
||||||
|
'dictionary',
|
||||||
|
'meetings',
|
||||||
|
'meeting_memos',
|
||||||
|
'meeting_documents',
|
||||||
|
'custom_instructions',
|
||||||
|
'user_templates',
|
||||||
|
'knowledge_documents'
|
||||||
|
]
|
||||||
|
LOOP
|
||||||
|
EXECUTE format('DROP TRIGGER IF EXISTS stamp_sync_insert_v1 ON public.%I', t);
|
||||||
|
EXECUTE format(
|
||||||
|
'CREATE TRIGGER stamp_sync_insert_v1 BEFORE INSERT ON public.%I '
|
||||||
|
'FOR EACH ROW EXECUTE FUNCTION public.stamp_sync_timestamp_v1()',
|
||||||
|
t
|
||||||
|
);
|
||||||
|
END LOOP;
|
||||||
|
END $$;
|
||||||
|
|
||||||
|
DROP TRIGGER IF EXISTS stamp_sync_update_v1 ON public.user_templates;
|
||||||
|
CREATE TRIGGER stamp_sync_update_v1
|
||||||
|
BEFORE UPDATE ON public.user_templates
|
||||||
|
FOR EACH ROW EXECUTE FUNCTION public.stamp_sync_timestamp_v1();
|
||||||
|
|
||||||
|
COMMIT;
|
||||||
|
|
@ -0,0 +1,197 @@
|
||||||
|
\set ON_ERROR_STOP on
|
||||||
|
|
||||||
|
-- Regression for 20260927000037_server_owned_sync_timestamps.sql.
|
||||||
|
-- Every synced table is pulled with a (updated_at, id) keyset cursor, so an
|
||||||
|
-- INSERT must not keep a client-chosen updated_at: a stale value (offline
|
||||||
|
-- recording, archive restore) lands behind other desktops' cursors, and a
|
||||||
|
-- future value (fast clock) pushes their cursors ahead.
|
||||||
|
|
||||||
|
BEGIN;
|
||||||
|
|
||||||
|
CREATE OR REPLACE FUNCTION pg_temp.assert_true(condition boolean, message text)
|
||||||
|
RETURNS void
|
||||||
|
LANGUAGE plpgsql
|
||||||
|
AS $$
|
||||||
|
BEGIN
|
||||||
|
IF condition IS NOT TRUE THEN
|
||||||
|
RAISE EXCEPTION 'assertion_failed: %', message;
|
||||||
|
END IF;
|
||||||
|
END;
|
||||||
|
$$;
|
||||||
|
|
||||||
|
-- Catalog: every cursor-pulled table stamps updated_at on INSERT.
|
||||||
|
DO $$
|
||||||
|
DECLARE
|
||||||
|
t text;
|
||||||
|
BEGIN
|
||||||
|
FOREACH t IN ARRAY ARRAY[
|
||||||
|
'history',
|
||||||
|
'dictionary',
|
||||||
|
'meetings',
|
||||||
|
'meeting_memos',
|
||||||
|
'meeting_documents',
|
||||||
|
'custom_instructions',
|
||||||
|
'user_templates',
|
||||||
|
'knowledge_documents'
|
||||||
|
]
|
||||||
|
LOOP
|
||||||
|
PERFORM pg_temp.assert_true(
|
||||||
|
EXISTS (
|
||||||
|
SELECT 1
|
||||||
|
FROM pg_trigger AS trg
|
||||||
|
WHERE trg.tgrelid = format('public.%I', t)::regclass
|
||||||
|
AND trg.tgname = 'stamp_sync_insert_v1'
|
||||||
|
AND NOT trg.tgisinternal
|
||||||
|
AND trg.tgenabled <> 'D'
|
||||||
|
-- ROW (1) | BEFORE (2) | INSERT (4)
|
||||||
|
AND (trg.tgtype & 7) = 7
|
||||||
|
),
|
||||||
|
format('%s stamps updated_at before insert', t)
|
||||||
|
);
|
||||||
|
END LOOP;
|
||||||
|
END $$;
|
||||||
|
|
||||||
|
SELECT pg_temp.assert_true(
|
||||||
|
EXISTS (
|
||||||
|
SELECT 1
|
||||||
|
FROM pg_trigger AS trg
|
||||||
|
WHERE trg.tgrelid = 'public.user_templates'::regclass
|
||||||
|
AND trg.tgname = 'stamp_sync_update_v1'
|
||||||
|
-- ROW (1) | BEFORE (2) | UPDATE (16)
|
||||||
|
AND (trg.tgtype & 19) = 19
|
||||||
|
),
|
||||||
|
'user_templates stamps updated_at before update'
|
||||||
|
);
|
||||||
|
|
||||||
|
SELECT pg_temp.assert_true(
|
||||||
|
NOT has_function_privilege('authenticated', 'public.stamp_sync_timestamp_v1()', 'EXECUTE'),
|
||||||
|
'clients cannot call the stamp function directly'
|
||||||
|
);
|
||||||
|
|
||||||
|
INSERT INTO auth.users (
|
||||||
|
id, aud, role, email, encrypted_password, email_confirmed_at,
|
||||||
|
raw_app_meta_data, raw_user_meta_data, created_at, updated_at
|
||||||
|
) VALUES (
|
||||||
|
'37000000-0000-4000-8000-000000000001', 'authenticated', 'authenticated',
|
||||||
|
'sync-timestamps@example.invalid', crypt('fixture-password', gen_salt('bf')), now(),
|
||||||
|
'{"provider":"email","providers":["email"]}'::jsonb, '{}'::jsonb, now(), now()
|
||||||
|
);
|
||||||
|
|
||||||
|
-- Behaviour: client-supplied past and future timestamps are replaced by the
|
||||||
|
-- transaction clock on every table.
|
||||||
|
INSERT INTO public.history (id, user_id, original_text, duration, updated_at)
|
||||||
|
VALUES
|
||||||
|
('37000000-0000-4000-8000-0000000000a1', '37000000-0000-4000-8000-000000000001',
|
||||||
|
'recorded offline yesterday', 1.5, now() - interval '1 day'),
|
||||||
|
('37000000-0000-4000-8000-0000000000a2', '37000000-0000-4000-8000-000000000001',
|
||||||
|
'recorded on a fast clock', 1.5, now() + interval '2 hours');
|
||||||
|
|
||||||
|
INSERT INTO public.dictionary (id, user_id, word, updated_at)
|
||||||
|
VALUES ('37000000-0000-4000-8000-0000000000b1', '37000000-0000-4000-8000-000000000001',
|
||||||
|
'cursor', now() - interval '3 days');
|
||||||
|
|
||||||
|
INSERT INTO public.meetings (id, user_id, updated_at)
|
||||||
|
VALUES ('37000000-0000-4000-8000-0000000000c1', '37000000-0000-4000-8000-000000000001',
|
||||||
|
now() + interval '5 hours');
|
||||||
|
|
||||||
|
INSERT INTO public.meeting_memos (id, meeting_id, user_id, content, timestamp_ms, updated_at)
|
||||||
|
VALUES ('37000000-0000-4000-8000-0000000000d1', '37000000-0000-4000-8000-0000000000c1',
|
||||||
|
'37000000-0000-4000-8000-000000000001', 'memo', 1000, now() - interval '10 days');
|
||||||
|
|
||||||
|
INSERT INTO public.meeting_documents (id, meeting_id, user_id, template_type, title, updated_at)
|
||||||
|
VALUES ('37000000-0000-4000-8000-0000000000e1', '37000000-0000-4000-8000-0000000000c1',
|
||||||
|
'37000000-0000-4000-8000-000000000001', 'minutes', 'minutes', now() - interval '1 year');
|
||||||
|
|
||||||
|
INSERT INTO public.custom_instructions (id, user_id, name, prompt, updated_at)
|
||||||
|
VALUES ('37000000-0000-4000-8000-0000000000f1', '37000000-0000-4000-8000-000000000001',
|
||||||
|
'sync timestamp fixture', 'prompt', now() + interval '1 day');
|
||||||
|
|
||||||
|
INSERT INTO public.user_templates (
|
||||||
|
id, user_id, template_kind, name, template_type, system_prompt, updated_at
|
||||||
|
) VALUES (
|
||||||
|
'37000000-0000-4000-8000-000000000101', '37000000-0000-4000-8000-000000000001',
|
||||||
|
'meeting_document', 'sync timestamp fixture', 'custom', 'prompt', now() - interval '30 days'
|
||||||
|
);
|
||||||
|
|
||||||
|
INSERT INTO public.knowledge_documents (id, user_id, title, updated_at)
|
||||||
|
VALUES ('37000000-0000-4000-8000-000000000111', '37000000-0000-4000-8000-000000000001',
|
||||||
|
'knowledge', now() - interval '2 days');
|
||||||
|
|
||||||
|
DO $$
|
||||||
|
DECLARE
|
||||||
|
owner uuid := '37000000-0000-4000-8000-000000000001';
|
||||||
|
t text;
|
||||||
|
stale integer;
|
||||||
|
BEGIN
|
||||||
|
FOREACH t IN ARRAY ARRAY[
|
||||||
|
'history',
|
||||||
|
'dictionary',
|
||||||
|
'meetings',
|
||||||
|
'meeting_memos',
|
||||||
|
'meeting_documents',
|
||||||
|
'custom_instructions',
|
||||||
|
'user_templates',
|
||||||
|
'knowledge_documents'
|
||||||
|
]
|
||||||
|
LOOP
|
||||||
|
EXECUTE format(
|
||||||
|
'SELECT count(*) FROM public.%I WHERE user_id = $1 AND id::text LIKE ''37000000-%%'' '
|
||||||
|
'AND updated_at IS DISTINCT FROM now()',
|
||||||
|
t
|
||||||
|
)
|
||||||
|
INTO stale
|
||||||
|
USING owner;
|
||||||
|
PERFORM pg_temp.assert_true(stale = 0, format('%s insert keeps the server clock, not the client value', t));
|
||||||
|
END LOOP;
|
||||||
|
END $$;
|
||||||
|
|
||||||
|
-- The desktop pushes with PostgREST upsert (INSERT ... ON CONFLICT DO UPDATE).
|
||||||
|
-- A fresh row goes through the INSERT stamp, an existing row through the
|
||||||
|
-- UPDATE stamp; neither keeps the client timestamp.
|
||||||
|
INSERT INTO public.history (id, user_id, original_text, duration, updated_at)
|
||||||
|
VALUES ('37000000-0000-4000-8000-0000000000a3', '37000000-0000-4000-8000-000000000001',
|
||||||
|
'upserted while offline', 2.0, now() - interval '6 hours')
|
||||||
|
ON CONFLICT (id) DO UPDATE SET original_text = EXCLUDED.original_text, updated_at = EXCLUDED.updated_at;
|
||||||
|
|
||||||
|
INSERT INTO public.history (id, user_id, original_text, duration, updated_at)
|
||||||
|
VALUES ('37000000-0000-4000-8000-0000000000a1', '37000000-0000-4000-8000-000000000001',
|
||||||
|
'edited offline', 1.5, now() - interval '6 hours')
|
||||||
|
ON CONFLICT (id) DO UPDATE SET original_text = EXCLUDED.original_text, updated_at = EXCLUDED.updated_at;
|
||||||
|
|
||||||
|
SELECT pg_temp.assert_true(
|
||||||
|
(SELECT count(*) FROM public.history
|
||||||
|
WHERE id IN ('37000000-0000-4000-8000-0000000000a1', '37000000-0000-4000-8000-0000000000a3')
|
||||||
|
AND updated_at = now()) = 2,
|
||||||
|
'upsert insert and upsert update both carry the server clock'
|
||||||
|
);
|
||||||
|
|
||||||
|
-- Same stamp for a client write through RLS, as PostgREST performs it.
|
||||||
|
SET LOCAL ROLE authenticated;
|
||||||
|
SELECT set_config(
|
||||||
|
'request.jwt.claims',
|
||||||
|
'{"sub":"37000000-0000-4000-8000-000000000001","role":"authenticated"}',
|
||||||
|
true
|
||||||
|
);
|
||||||
|
|
||||||
|
INSERT INTO public.dictionary (id, user_id, word, updated_at)
|
||||||
|
VALUES ('37000000-0000-4000-8000-0000000000b2', '37000000-0000-4000-8000-000000000001',
|
||||||
|
'skewed', now() + interval '2 hours');
|
||||||
|
|
||||||
|
SELECT pg_temp.assert_true(
|
||||||
|
(SELECT updated_at FROM public.dictionary WHERE id = '37000000-0000-4000-8000-0000000000b2') = now(),
|
||||||
|
'an authenticated client cannot choose updated_at on insert'
|
||||||
|
);
|
||||||
|
|
||||||
|
RESET ROLE;
|
||||||
|
|
||||||
|
-- user_templates: a direct UPDATE (no RPC) also moves the cursor column.
|
||||||
|
UPDATE public.user_templates
|
||||||
|
SET name = 'renamed fixture', updated_at = now() - interval '9 days'
|
||||||
|
WHERE id = '37000000-0000-4000-8000-000000000101';
|
||||||
|
|
||||||
|
SELECT pg_temp.assert_true(
|
||||||
|
(SELECT updated_at FROM public.user_templates WHERE id = '37000000-0000-4000-8000-000000000101') = now(),
|
||||||
|
'user_templates update carries the server clock'
|
||||||
|
);
|
||||||
|
|
||||||
|
ROLLBACK;
|
||||||
Loading…
Add table
Add a link
Reference in a new issue