diff --git a/synapse/storage/databases/main/task_scheduler.py b/synapse/storage/databases/main/task_scheduler.py
index 1fb3180c3c..9ab120eea9 100644
--- a/synapse/storage/databases/main/task_scheduler.py
+++ b/synapse/storage/databases/main/task_scheduler.py
@@ -92,7 +92,7 @@ class TaskSchedulerWorkerStore(SQLBaseStore):
if clauses:
sql = sql + " WHERE " + " AND ".join(clauses)
- sql = sql + "ORDER BY timestamp"
+ sql = sql + " ORDER BY timestamp"
txn.execute(sql, args)
return self.db_pool.cursor_to_dict(txn)
diff --git a/synapse/storage/databases/main/transactions.py b/synapse/storage/databases/main/transactions.py
index 48e4b0ba3c..860bbf7c0f 100644
--- a/synapse/storage/databases/main/transactions.py
+++ b/synapse/storage/databases/main/transactions.py
@@ -242,8 +242,6 @@ class TransactionWorkerStore(CacheInvalidationWorkerStore):
) -> None:
# Upsert retry time interval if retry_interval is zero (i.e. we're
# resetting it) or greater than the existing retry interval.
- # We also upsert when the new retry interval is the same as the existing one,
- # since it will be the case when `destination_max_retry_interval` is reached.
#
# WARNING: This is executed in autocommit, so we shouldn't add any more
# SQL calls in here (without being very careful).
@@ -258,8 +256,10 @@ class TransactionWorkerStore(CacheInvalidationWorkerStore):
retry_interval = EXCLUDED.retry_interval
WHERE
EXCLUDED.retry_interval = 0
+ OR EXCLUDED.retry_last_ts = 0
OR destinations.retry_interval IS NULL
- OR destinations.retry_interval <= EXCLUDED.retry_interval
+ OR destinations.retry_interval < EXCLUDED.retry_interval
+ OR destinations.retry_last_ts < EXCLUDED.retry_last_ts
"""
txn.execute(sql, (destination, failure_ts, retry_last_ts, retry_interval))
diff --git a/synapse/storage/schema/main/delta/80/02_read_write_locks_deadlock.sql.postgres b/synapse/storage/schema/main/delta/80/02_read_write_locks_deadlock.sql.postgres
new file mode 100644
index 0000000000..401c42e18a
--- /dev/null
+++ b/synapse/storage/schema/main/delta/80/02_read_write_locks_deadlock.sql.postgres
@@ -0,0 +1,37 @@
+/* Copyright 2023 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+-- To avoid the possibility of a deadlock, lock the
+-- `worker_read_write_locks_mode` table so that we serialize inserts/deletes
+-- for a specific lock name/key.
+
+CREATE OR REPLACE FUNCTION delete_read_write_lock_parent_before() RETURNS trigger AS $$
+BEGIN
+ -- `PERFORM` is a `SELECT` which discards the rows.
+ PERFORM * FROM worker_read_write_locks_mode
+ WHERE
+ lock_name = OLD.lock_name
+ AND lock_key = OLD.lock_key
+ FOR UPDATE;
+
+ RETURN OLD;
+END
+$$
+LANGUAGE plpgsql;
+
+DROP TRIGGER IF EXISTS delete_read_write_lock_parent_before_trigger ON worker_read_write_locks;
+CREATE TRIGGER delete_read_write_lock_parent_before_trigger BEFORE DELETE ON worker_read_write_locks
+ FOR EACH ROW
+ EXECUTE PROCEDURE delete_read_write_lock_parent_before();
diff --git a/synapse/storage/schema/main/delta/80/03_read_write_locks_triggers.sql.postgres b/synapse/storage/schema/main/delta/80/03_read_write_locks_triggers.sql.postgres
new file mode 100644
index 0000000000..31de5bfa18
--- /dev/null
+++ b/synapse/storage/schema/main/delta/80/03_read_write_locks_triggers.sql.postgres
@@ -0,0 +1,37 @@
+/* Copyright 2023 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+-- Fix up the triggers that were in `78/04_read_write_locks_triggers.sql`
+
+-- Reduce the number of writes we do on this table.
+--
+-- Note: that we still want to lock the row here (i.e. still do a `DO UPDATE
+-- SET`) so that we serialize updates.
+CREATE OR REPLACE FUNCTION upsert_read_write_lock_parent() RETURNS trigger AS $$
+BEGIN
+ INSERT INTO worker_read_write_locks_mode (lock_name, lock_key, write_lock, token)
+ VALUES (NEW.lock_name, NEW.lock_key, NEW.write_lock, NEW.token)
+ ON CONFLICT (lock_name, lock_key)
+ DO UPDATE SET write_lock = NEW.write_lock
+ WHERE OLD.write_lock != NEW.write_lock;
+ RETURN NEW;
+END
+$$
+LANGUAGE plpgsql;
+
+DROP TRIGGER IF EXISTS upsert_read_write_lock_parent_trigger ON worker_read_write_locks;
+CREATE TRIGGER upsert_read_write_lock_parent_trigger BEFORE INSERT ON worker_read_write_locks
+ FOR EACH ROW
+ EXECUTE PROCEDURE upsert_read_write_lock_parent();
diff --git a/synapse/util/task_scheduler.py b/synapse/util/task_scheduler.py
index 773a8327f6..4aea64b338 100644
--- a/synapse/util/task_scheduler.py
+++ b/synapse/util/task_scheduler.py
@@ -154,13 +154,15 @@ class TaskScheduler:
f"No function associated with action {action} of the scheduled task"
)
+ status = TaskStatus.SCHEDULED
if timestamp is None or timestamp < self._clock.time_msec():
timestamp = self._clock.time_msec()
+ status = TaskStatus.ACTIVE
task = ScheduledTask(
random_string(16),
action,
- TaskStatus.SCHEDULED,
+ status,
timestamp,
resource_id,
params,
|
