fix: publish grid cache cells atomically and preserve invalidations

django/applications/catmaid/control/grid_cache.py

@@ -0,0 +1,66 @@
+"""Coordinate concurrent grid cell refreshes and publish each cell atomically."""
+
+from contextlib import contextmanager
+import hashlib
+
+
+def cell_lock_id(grid_id, x, y, z):
+    """Give cooperating writers a stable transaction advisory lock per cell."""
+    value = f'catmaid.grid-cell:{grid_id}:{x}:{y}:{z}'.encode('ascii')
+    return int.from_bytes(hashlib.sha256(value).digest()[:8], 'big', signed=True)
+
+
+@contextmanager
+def locked_grid_cell(grid_id, x, y, z, *, connection=None,
+        atomic=None, lock_timeout_ms=None, statement_timeout_ms=None):
+    """Publish all formats atomically and acknowledge only an observed invalidation.
+
+    Spatial notifications may arrive while a cell is being generated. Compare
+    the dirty row's MVCC version, not just its timestamp (``now()`` is the
+    invalidator's transaction start time), so a newer invalidation survives.
+
+    Cache workers and other refresh callers must acquire the lock before reading
+    source data. This prevents an older calculation from overwriting a newer
+    result. Readers see the previous committed payload until every format and
+    the invalidation acknowledgement commit together. Writers that do not use
+    this helper do not participate in its advisory locking protocol.
+    """
+    if connection is None or atomic is None:
+        from django.db import connection as default_connection, transaction
+        connection = connection if connection is not None else default_connection
+        atomic = atomic or (lambda: transaction.atomic(using=connection.alias))
+    # Every format and its invalidation acknowledgement must share one cell's
+    # commit or rollback, independently of a caller's longer-running operation.
+    if connection.in_atomic_block:
+        raise RuntimeError('Grid cell refresh requires its own transaction')
+    coordinates = [grid_id, x, y, z]
+    with atomic(), connection.cursor() as cursor:
+        for name, timeout in (('lock_timeout', lock_timeout_ms),
+                ('statement_timeout', statement_timeout_ms)):
+            if timeout is not None:
+                cursor.execute('SELECT set_config(%s, %s, true)',
+                        [name, f'{int(timeout)}ms'])
+        cursor.execute('SELECT pg_advisory_xact_lock(%s)',
+                [cell_lock_id(*coordinates)])
+        cursor.execute('''
+            SELECT id, xmin::text FROM dirty_node_grid_cache_cell
+            WHERE grid_id=%s AND x_index=%s AND y_index=%s AND z_index=%s
+        ''', coordinates)
+        invalidation = cursor.fetchone()
+        cursor.execute('SELECT clock_timestamp()')
+        started_at = cursor.fetchone()[0]
+        receipt = {'generated_at': started_at, 'present': False}
+        yield cursor, receipt
+        # update_grid_cell uses now(), which can predate waiting for the lock.
+        # Record the actual query barrier, shared by every payload format.
+        cursor.execute('''
+            UPDATE node_grid_cache_cell SET update_time=%s
+            WHERE grid_id=%s AND x_index=%s AND y_index=%s AND z_index=%s
+            RETURNING id
+        ''', [started_at, *coordinates])
+        receipt['present'] = cursor.fetchone() is not None
+        if invalidation is not None:
+            cursor.execute('''
+                DELETE FROM dirty_node_grid_cache_cell
+                WHERE id=%s AND xmin::text=%s
+            ''', list(invalidation))

django/applications/catmaid/management/commands/catmaid_cache_update_worker.py

@@ -13,9 +13,10 @@
 
 from catmaid.consumers import msg_all
 from catmaid.control.edge import get_intersected_grid_cells
+from catmaid.control.grid_cache import locked_grid_cell
 from catmaid.control.node import (get_configured_node_providers,
         GridCachedNodeProvider, Postgis3dNodeProvider, update_grid_cell)
-from catmaid.models import NodeGridCache, DirtyNodeGridCacheCell
+from catmaid.models import NodeGridCache
 from catmaid.util import str2bool
 from .common import set_log_level
 
@@ -41,7 +42,6 @@ def update(self, updates, cursor):
         grid_map = dict((g.id, g) for g in grids)
 
         provider = Postgis3dNodeProvider()
-        cursor = connection.cursor()
 
         # Batch of grids to update during one run
         updated_cells = 0
@@ -63,12 +63,15 @@ def update(self, updates, cursor):
             if g.hidden_last_editor_id:
                 params['hidden_last_editor_id'] = int(g.hidden_last_editor_id)
 
-            updated = update_grid_cell(g.project_id, g.id, w_i, h_i, d_i,
-                    g.cell_width, g.cell_height, g.cell_depth,
-                    params, g.allow_empty, g.n_lod_levels, g.lod_min_bucket_size,
-                    g.lod_strategy, g.has_json_data, g.has_json_text_data,
-                    g.has_msgpack_data, provider=provider, cursor=cursor,
-                    delete_empty=True)
+            # Lock before querying source data; publish every format and clear
+            # only the invalidation observed at the start of this refresh.
+            with locked_grid_cell(g.id, w_i, h_i, d_i) as (cell_cursor, _receipt):
+                updated = update_grid_cell(g.project_id, g.id, w_i, h_i, d_i,
+                        g.cell_width, g.cell_height, g.cell_depth,
+                        params, g.allow_empty, g.n_lod_levels, g.lod_min_bucket_size,
+                        g.lod_strategy, g.has_json_data, g.has_json_text_data,
+                        g.has_msgpack_data, provider=provider, cursor=cell_cursor,
+                        delete_empty=True)
 
             if updated:
                 updated_cells += 1
@@ -81,10 +84,6 @@ def update(self, updates, cursor):
                     'z': d_i,
                 }])
 
-                # TODO: delete in batches
-                DirtyNodeGridCacheCell.objects.filter(grid_id=g.id, x_index=w_i,
-                        y_index=h_i, z_index=d_i).delete()
-
         # Optionally, let users know about updated cells
         if settings.SPATIAL_UPDATE_CLIENT_NOTIFICATIONS:
             msg_all('grid-cache-update', {