Skip to content

Commit e99674e

Browse files
maparentmaparent
committed
locallly applied plpgsql_check to plpgsql functions
1 parent d141800 commit e99674e

4 files changed

Lines changed: 103 additions & 9 deletions

File tree

‎packages/database/supabase/migrations/20250517154122_plpgsql_linting.sql‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
CREATE OR REPLACE FUNCTION public.end_sync_task(s_target bigint, s_function character varying, s_worker character varying, s_status task_status)
2+
RETURNS void
3+
LANGUAGE plpgsql
4+
AS $function$
5+
DECLARE t_id INTEGER;
6+
DECLARE t_worker varchar;
7+
DECLARE t_status task_status;
8+
DECLARE t_failure_count SMALLINT;
9+
DECLARE t_last_task_end TIMESTAMP WITH TIME ZONE;
10+
BEGIN
11+
SELECT id, worker, status, failure_count, last_task_end
12+
INTO STRICT t_id, t_worker, t_status, t_failure_count, t_last_task_end
13+
FROM sync_info WHERE sync_target = s_target AND sync_function = s_function;
14+
ASSERT s_status > 'active';
15+
ASSERT t_worker = s_worker, 'Wrong worker';
16+
ASSERT s_status >= t_status, 'do not go back in status';
17+
IF s_status = 'complete' THEN
18+
t_last_task_end := now();
19+
t_failure_count := 0;
20+
ELSE
21+
IF t_status != s_status THEN
22+
t_failure_count := t_failure_count + 1;
23+
END IF;
24+
END IF;
25+
26+
UPDATE sync_info
27+
SET status = s_status,
28+
task_times_out_at=null,
29+
last_task_end=t_last_task_end,
30+
failure_count=t_failure_count
31+
WHERE id=t_id;
32+
END;
33+
$function$
34+
;
35+
36+
CREATE OR REPLACE FUNCTION public.propose_sync_task(s_target bigint, s_function character varying, s_worker character varying, timeout interval, task_interval interval)
37+
RETURNS interval
38+
LANGUAGE plpgsql
39+
AS $function$
40+
DECLARE s_id INTEGER;
41+
DECLARE start_time TIMESTAMP WITH TIME ZONE;
42+
DECLARE t_status task_status;
43+
DECLARE t_failure_count SMALLINT;
44+
DECLARE t_last_task_start TIMESTAMP WITH TIME ZONE;
45+
DECLARE t_last_task_end TIMESTAMP WITH TIME ZONE;
46+
DECLARE t_times_out_at TIMESTAMP WITH TIME ZONE;
47+
DECLARE result INTERVAL = NULL;
48+
BEGIN
49+
ASSERT timeout * 2 < task_interval;
50+
ASSERT timeout >= '1s'::interval;
51+
ASSERT task_interval >= '5s'::interval;
52+
start_time := now();
53+
INSERT INTO sync_info (sync_target, sync_function, status, worker, last_task_start, task_times_out_at)
54+
VALUES (s_target, s_function, 'active', s_worker, start_time, start_time+timeout)
55+
ON CONFLICT DO NOTHING
56+
RETURNING id INTO s_id;
57+
-- zut il renvoie null...
58+
IF s_id IS NOT NULL THEN
59+
-- totally new_row, I'm on the task
60+
RETURN NULL;
61+
END IF;
62+
-- now we know it pre-existed. Maybe already active.
63+
SELECT id INTO STRICT s_id FROM sync_info WHERE sync_target = s_target AND sync_function = s_function;
64+
PERFORM pg_advisory_lock(s_id);
65+
SELECT status, failure_count, last_task_start, last_task_end, task_times_out_at
66+
INTO t_status, t_failure_count, t_last_task_start, t_last_task_end, t_times_out_at
67+
FROM sync_info
68+
WHERE id = s_id;
69+
70+
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
71+
t_status := 'timeout';
72+
t_failure_count := t_failure_count + 1;
73+
END IF;
74+
-- basic backoff
75+
task_interval := task_interval * (1+t_failure_count);
76+
IF coalesce(t_last_task_end, t_last_task_start) + task_interval < now() THEN
77+
-- we are ready to take on the task
78+
UPDATE sync_info
79+
SET worker=s_worker, status='active', task_times_out_at = now() + timeout, last_task_start = now(), failure_count=t_failure_count
80+
WHERE id=s_id;
81+
ELSE
82+
-- the task has been tried recently enough
83+
IF t_status = 'timeout' THEN
84+
UPDATE sync_info
85+
SET status=t_status, failure_count=t_failure_count
86+
WHERE id=s_id;
87+
END IF;
88+
result := coalesce(t_last_task_end, t_last_task_start) + task_interval - now();
89+
END IF;
90+
91+
PERFORM pg_advisory_unlock(s_id);
92+
RETURN result;
93+
END;
94+
$function$
95+
;

‎packages/database/supabase/schemas/base.sql‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ SET idle_in_transaction_session_timeout = 0;
66
SET client_encoding = 'UTF8';
77
SET standard_conforming_strings = on;
88
SELECT pg_catalog.set_config('search_path', '', false);
9-
SET check_function_bodies = false;
9+
SET check_function_bodies = true;
1010
SET xmloption = content;
1111
SET client_min_messages = warning;
1212
SET row_security = on;

‎packages/database/supabase/schemas/embedding.sql‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ GRANT ALL ON TABLE "public"."ContentEmbedding_openai_text_embedding_3_small_1536
2828
GRANT ALL ON TABLE "public"."ContentEmbedding_openai_text_embedding_3_small_1536" TO "authenticated";
2929
GRANT ALL ON TABLE "public"."ContentEmbedding_openai_text_embedding_3_small_1536" TO "service_role";
3030

31-
31+
set search_path to public,extensions;
3232

3333
CREATE OR REPLACE FUNCTION "public"."match_content_embeddings"("query_embedding" "extensions"."vector", "match_threshold" double precision, "match_count" integer, "current_document_id" integer DEFAULT NULL::integer) RETURNS TABLE("content_id" bigint, "roam_uid" "text", "text_content" "text", "similarity" double precision)
3434
LANGUAGE "sql" STABLE
@@ -75,3 +75,5 @@ ORDER BY similarity DESC; -- Order by calculated similarity, highest first
7575
$$;
7676

7777
ALTER FUNCTION "public"."match_embeddings_for_subset_nodes"("p_query_embedding" "extensions"."vector", "p_subset_roam_uids" "text"[]) OWNER TO "postgres";
78+
79+
set search_path to '';

‎packages/database/supabase/schemas/sync.sql‎

Lines changed: 4 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,6 @@ CREATE OR REPLACE FUNCTION "public"."end_sync_task"("s_target" bigint, "s_functi
4747
LANGUAGE "plpgsql"
4848
AS $$
4949
DECLARE t_id INTEGER;
50-
DECLARE t_target varchar;
5150
DECLARE t_worker varchar;
5251
DECLARE t_status task_status;
5352
DECLARE t_failure_count SMALLINT;
@@ -57,8 +56,8 @@ BEGIN
5756
INTO STRICT t_id, t_worker, t_status, t_failure_count, t_last_task_end
5857
FROM sync_info WHERE sync_target = s_target AND sync_function = s_function;
5958
ASSERT s_status > 'active';
60-
ASSERT t_worker = s_worker, "Wrong worker";
61-
ASSERT s_status >= t_status, "do not go back in status";
59+
ASSERT t_worker = s_worker, 'Wrong worker';
60+
ASSERT s_status >= t_status, 'do not go back in status';
6261
IF s_status = 'complete' THEN
6362
t_last_task_end := now();
6463
t_failure_count := 0;
@@ -84,9 +83,7 @@ CREATE OR REPLACE FUNCTION "public"."propose_sync_task"("s_target" bigint, "s_fu
8483
LANGUAGE "plpgsql"
8584
AS $$
8685
DECLARE s_id INTEGER;
87-
DECLARE timeout_as TIMESTAMP WITH TIME ZONE;
8886
DECLARE start_time TIMESTAMP WITH TIME ZONE;
89-
DECLARE t_worker VARCHAR;
9087
DECLARE t_status task_status;
9188
DECLARE t_failure_count SMALLINT;
9289
DECLARE t_last_task_start TIMESTAMP WITH TIME ZONE;
@@ -110,8 +107,8 @@ BEGIN
110107
-- now we know it pre-existed. Maybe already active.
111108
SELECT id INTO STRICT s_id FROM sync_info WHERE sync_target = s_target AND sync_function = s_function;
112109
PERFORM pg_advisory_lock(s_id);
113-
SELECT worker, status, failure_count, last_task_start, last_task_end, task_times_out_at
114-
INTO t_worker, t_status, t_failure_count, t_last_task_start, t_last_task_end, t_times_out_at
110+
SELECT status, failure_count, last_task_start, last_task_end, task_times_out_at
111+
INTO t_status, t_failure_count, t_last_task_start, t_last_task_end, t_times_out_at
115112
FROM sync_info
116113
WHERE id = s_id;
117114

0 commit comments

Comments
 (0)