Conditional locking in Postgres function/stored procedure

user_6950600 0
Viewed 479

I'm building an event sourcing service for a web crawler, where there are several crawler workers scraping several websites and trying to keep deltas for the crawled resource. I've chosen PostgreSQL for the underlying data store. I need to give producers the ability to have optimistic locking using a flag called "expectedSeq" to control whether or not the event should be written for a particular stream. Initially, I was using a table, leveraging the auto-increment with a transaction to build the optimistic locking feature for each "stream" but I quickly found there's a file system cap on how many tables a server can handle.

Since I can't use auto-increment anymore, I'm trying to build this functionality using two tables, one for controlling the sequence of the stream, the other for storing the event itself.

The first question I have is, whether I should use store procedures or functions. The second is it possible to have conditional transactions inside a stored procedure or a Postgres function.

The logic I need to implemented is something of sorts

storeEvent(stream, expectedSeq = null)

lock row for `streams`.stream

if expectedSeq = null
  update stream row with seq + 1
  release lock
  write event to event table
else
  if expectedSeq != seq + 1
    release lock
    abort
  else
    update seq + 1
    release lock
    write event to event table
postgresql
1 Answers

thanks to Ian Harris

CREATE OR REPLACE PROCEDURE store_event (v_topic varchar(40), v_expected_next_seq integer, v_data text)
LANGUAGE plpgsql
AS $$
DECLARE
  next_seq integer;
BEGIN
  -- FOR UPDATE clause places row level lock on table
  next_seq := (
    SELECT
      seq
    FROM
      topics
    WHERE
      topic = v_topic
    FOR UPDATE) + 1;
  IF v_expected_next_seq IS NOT NULL AND next_seq != v_expected_next_seq THEN
    RAISE 'Optimistic locking error';
  END IF;
  IF next_seq IS NULL THEN
    RAISE 'Unknown topic';
  END IF;
  UPDATE
    topics
  SET
    seq = next_seq
  WHERE
    topic = v_topic;
  INSERT INTO events (topic, seq, data)
    VALUES (v_topic, next_seq, v_data);
  COMMIT;
END;
$$;
user_69506000
Related