ENG-447 Adding last sync to sync task return value (#216)

* ENG-447 Adding last sync to sync task return value
Also incorporated updated documentation (ENG-443)

Author: maparent

packages/database/example.md renamed to packages/database/doc/concept_example.md

@@ -0,0 +1,18 @@
+# Sync information
+
+The `sync_info` table is meant to always be accessed through one of these two functions: `propose_sync_task` and `end_sync_task`.
+This acts as a semaphore, so that two workers (e.g. the roam plugin on two different browsers) do not try to run the same sync task at the same time. So you need to give the function `propose_sync_task` enough information to distinguish what you mean to do:
+
+1. The `target`, e.g. the database Id of the scope of the task, usually a space, but it could be a single content or concept (for reactive updates)
+2.  a `function` name, to distinguish different tasks on the same target; e.g. adding vs deleting content. (arbitrary short string)
+3.  the `worker` name: random string, should be the same between calls.
+
+Further, you may specify the `timeout` (>= 1s) after which the task should be deemed to have failed. The `task_interval` (>=5s) which is how often to do the task. (This must be longer than the `timeout`.)
+
+When a worker calls `propose_sync_task`, it will receive either:
+
+1. a timestamp in the future, meaning that the task is already being run by another worker, or has been run more recently than task_interval, and this worker should not attempt to run this task again before the given timestamp;
+2. a timestamp in the past, which is also the last time the task was executed successfully. Your worker can ask the platform for all changes posterior to that time. (This will only be reliable if you make always make your queries after calling this function!)
+3. Null, meaning the task was not executed successfully before, and your worker is tasked with starting from scratch.
+
+When a worker finishes the task, it should clean up with `end_sync_task`, giving the same identifying arguments and a status ('complete' or 'failed'.)

packages/database/doc/sync_functions.md

@@ -0,0 +1,107 @@
+DROP FUNCTION IF EXISTS public.propose_sync_task;
+
+CREATE OR REPLACE FUNCTION public.propose_sync_task(
+    s_target bigint,
+    s_function character varying,
+    s_worker character varying,
+    "timeout" interval,
+    "task_interval" interval
+) RETURNS timestamp with time zone
+LANGUAGE plpgsql
+AS $$
+DECLARE s_id INTEGER;
+DECLARE start_time TIMESTAMP WITH TIME ZONE := now();
+DECLARE t_status task_status;
+DECLARE t_failure_count SMALLINT;
+DECLARE t_last_task_start TIMESTAMP WITH TIME ZONE;
+DECLARE t_last_task_end TIMESTAMP WITH TIME ZONE;
+DECLARE t_times_out_at TIMESTAMP WITH TIME ZONE;
+DECLARE result TIMESTAMP WITH TIME ZONE;
+BEGIN
+    ASSERT timeout * 2 < task_interval;
+    ASSERT timeout >= '1s'::interval;
+    ASSERT task_interval >= '5s'::interval;
+    INSERT INTO public.sync_info (sync_target, sync_function, status, worker, last_task_start, task_times_out_at)
+        VALUES (s_target, s_function, 'active', s_worker, start_time, start_time+timeout)
+        ON CONFLICT (sync_target, sync_function) DO NOTHING
+        RETURNING id INTO s_id;
+    IF s_id IS NOT NULL THEN
+        -- totally new_row, I'm on the task
+        -- return last time it was run successfully
+        SELECT max(last_task_start) INTO result FROM public.sync_info
+            WHERE sync_target = s_target
+            AND sync_function = s_function
+            AND status = 'complete';
+        RETURN result;
+    END IF;
+    -- now we know it pre-existed. Maybe already active.
+    SELECT id INTO STRICT s_id FROM public.sync_info WHERE sync_target = s_target AND sync_function = s_function;
+    PERFORM pg_advisory_lock(s_id);
+    SELECT status, failure_count, last_task_start, last_task_end, task_times_out_at
+        INTO t_status, t_failure_count, t_last_task_start, t_last_task_end, t_times_out_at
+        FROM public.sync_info
+        WHERE id = s_id;
+
+    IF t_status = 'active' AND t_last_task_start >= coalesce(t_last_task_end, t_last_task_start) AND start_time > t_times_out_at THEN
+        t_status := 'timeout';
+        t_failure_count := t_failure_count + 1;
+    END IF;
+    -- basic backoff
+    task_interval := task_interval * (1+t_failure_count);
+    IF coalesce(t_last_task_end, t_last_task_start) + task_interval < now() THEN
+        -- we are ready to take on the task
+        UPDATE public.sync_info
+        SET worker=s_worker, status='active', task_times_out_at = now() + timeout, last_task_start = start_time, failure_count=t_failure_count
+        WHERE id=s_id;
+    ELSE
+        -- the task has been tried recently enough
+        IF t_status = 'timeout' THEN
+            UPDATE public.sync_info
+            SET status=t_status, failure_count=t_failure_count
+            WHERE id=s_id;
+        END IF;
+        result := coalesce(t_last_task_end, t_last_task_start) + task_interval;
+    END IF;
+
+    PERFORM pg_advisory_unlock(s_id);
+    RETURN result;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION public.end_sync_task(
+    s_target bigint,
+    s_function character varying,
+    s_worker character varying,
+    s_status public.task_status
+) RETURNS void
+LANGUAGE plpgsql
+AS $$
+DECLARE t_id INTEGER;
+DECLARE t_worker varchar;
+DECLARE t_status task_status;
+DECLARE t_failure_count SMALLINT;
+DECLARE t_last_task_end TIMESTAMP WITH TIME ZONE;
+BEGIN
+    SELECT id, worker, status, failure_count, last_task_end
+        INTO STRICT t_id, t_worker, t_status, t_failure_count, t_last_task_end
+        FROM public.sync_info WHERE sync_target = s_target AND sync_function = s_function;
+    ASSERT s_status > 'active';
+    ASSERT t_worker = s_worker, 'Wrong worker';
+    ASSERT s_status >= t_status, 'do not go back in status';
+    IF s_status = 'complete' THEN
+        t_last_task_end := now();
+        t_failure_count := 0;
+    ELSE
+        IF t_status != s_status THEN
+            t_failure_count := t_failure_count + 1;
+        END IF;
+    END IF;
+
+    UPDATE public.sync_info
+        SET status = s_status,
+            task_times_out_at=null,
+            last_task_end=t_last_task_end,
+            failure_count=t_failure_count
+        WHERE id=t_id;
+END;
+$$;

packages/database/supabase/migrations/20250616173959_extend_propose_sync_task.sql

@@ -44,9 +44,6 @@ CREATE UNIQUE INDEX sync_info_u_idx ON public.sync_info USING btree (
     "sync_target", sync_function
 );
 
-set search_path to public, extensions ;
-
-
 CREATE OR REPLACE FUNCTION public.end_sync_task(
     s_target bigint,
     s_function character varying,
@@ -63,7 +60,7 @@ DECLARE t_last_task_end TIMESTAMP WITH TIME ZONE;
 BEGIN
     SELECT id, worker, status, failure_count, last_task_end
         INTO STRICT t_id, t_worker, t_status, t_failure_count, t_last_task_end
-        FROM sync_info WHERE sync_target = s_target AND sync_function = s_function;
+        FROM public.sync_info WHERE sync_target = s_target AND sync_function = s_function;
     ASSERT s_status > 'active';
     ASSERT t_worker = s_worker, 'Wrong worker';
     ASSERT s_status >= t_status, 'do not go back in status';
@@ -76,7 +73,7 @@ BEGIN
         END IF;
     END IF;
 
-    UPDATE sync_info
+    UPDATE public.sync_info
         SET status = s_status,
             task_times_out_at=null,
             last_task_end=t_last_task_end,
@@ -99,37 +96,40 @@ CREATE OR REPLACE FUNCTION public.propose_sync_task(
     s_worker character varying,
     "timeout" interval,
     "task_interval" interval
-) RETURNS interval
+) RETURNS timestamp with time zone
 LANGUAGE plpgsql
 AS $$
 DECLARE s_id INTEGER;
-DECLARE start_time TIMESTAMP WITH TIME ZONE;
+DECLARE start_time TIMESTAMP WITH TIME ZONE := now();
 DECLARE t_status task_status;
 DECLARE t_failure_count SMALLINT;
 DECLARE t_last_task_start TIMESTAMP WITH TIME ZONE;
 DECLARE t_last_task_end TIMESTAMP WITH TIME ZONE;
 DECLARE t_times_out_at TIMESTAMP WITH TIME ZONE;
-DECLARE result INTERVAL = NULL;
+DECLARE result TIMESTAMP WITH TIME ZONE;
 BEGIN
     ASSERT timeout * 2 < task_interval;
     ASSERT timeout >= '1s'::interval;
     ASSERT task_interval >= '5s'::interval;
-    start_time := now();
-    INSERT INTO sync_info (sync_target, sync_function, status, worker, last_task_start, task_times_out_at)
+    INSERT INTO public.sync_info (sync_target, sync_function, status, worker, last_task_start, task_times_out_at)
         VALUES (s_target, s_function, 'active', s_worker, start_time, start_time+timeout)
-        ON CONFLICT DO NOTHING
+        ON CONFLICT (sync_target, sync_function) DO NOTHING
         RETURNING id INTO s_id;
-    -- zut il renvoie null...
     IF s_id IS NOT NULL THEN
         -- totally new_row, I'm on the task
-        RETURN NULL;
+        -- return last time it was run successfully
+        SELECT max(last_task_start) INTO result FROM public.sync_info
+            WHERE sync_target = s_target
+            AND sync_function = s_function
+            AND status = 'complete';
+        RETURN result;
     END IF;
     -- now we know it pre-existed. Maybe already active.
-    SELECT id INTO STRICT s_id FROM sync_info WHERE sync_target = s_target AND sync_function = s_function;
+    SELECT id INTO STRICT s_id FROM public.sync_info WHERE sync_target = s_target AND sync_function = s_function;
     PERFORM pg_advisory_lock(s_id);
     SELECT status, failure_count, last_task_start, last_task_end, task_times_out_at
         INTO t_status, t_failure_count, t_last_task_start, t_last_task_end, t_times_out_at
-        FROM sync_info
+        FROM public.sync_info
         WHERE id = s_id;
 
     IF t_status = 'active' AND t_last_task_start >= coalesce(t_last_task_end, t_last_task_start) AND start_time > t_times_out_at THEN
@@ -140,17 +140,17 @@ BEGIN
     task_interval := task_interval * (1+t_failure_count);
     IF coalesce(t_last_task_end, t_last_task_start) + task_interval < now() THEN
         -- we are ready to take on the task
-        UPDATE sync_info
-        SET worker=s_worker, status='active', task_times_out_at = now() + timeout, last_task_start = now(), failure_count=t_failure_count
+        UPDATE public.sync_info
+        SET worker=s_worker, status='active', task_times_out_at = now() + timeout, last_task_start = start_time, failure_count=t_failure_count
         WHERE id=s_id;
     ELSE
         -- the task has been tried recently enough
         IF t_status = 'timeout' THEN
-            UPDATE sync_info
+            UPDATE public.sync_info
             SET status=t_status, failure_count=t_failure_count
             WHERE id=s_id;
         END IF;
-        result := coalesce(t_last_task_end, t_last_task_start) + task_interval - now();
+        result := coalesce(t_last_task_end, t_last_task_start) + task_interval;
     END IF;
 
     PERFORM pg_advisory_unlock(s_id);