Clean up package pins when deposits roll back

src/msg/core/contracts.py

@@ -3,6 +3,7 @@
 from .models import *
 
 class MetadataSession(Protocol):
+    def on_rollback(self, effect: Callable[[], Awaitable[None]]) -> None: ...
     async def resource(self, id: ResourceId) -> Resource: ...
     async def resolve(self, path: str) -> ResourceId: ...
     async def children(

src/msg/plugins/store.py

@@ -210,6 +210,7 @@ async def package_deposit(ctx, request, tx):
                   'total_size':total,'delivery_mode':'managed_instant'}
         package_digest=digest(snapshot)
         for blob in blobs:
+            tx.on_rollback(lambda blob=blob: app.contents.unpin(blob, package_revision))
             await app.contents.pin(blob, package_revision)
         tx.execute('''INSERT INTO store_packages
             (id,listing_id,listing_revision,seller,revision,kind,manifest,payload_refs,

src/msg/storage/postgres.py

@@ -385,13 +385,15 @@ async def transaction(self, *, write):
             name = 'nested_' + uuid.uuid4().hex
             existing.execute('SAVEPOINT ' + name)
             pending_at_entry = len(existing.pending_effect_ids)
+            rollback_at = len(existing.rollback_effects)
             try:
                 yield existing
                 existing.execute('RELEASE SAVEPOINT ' + name)
             except BaseException:
                 existing.execute('ROLLBACK TO SAVEPOINT ' + name)
                 existing.execute('RELEASE SAVEPOINT ' + name)
                 del existing.pending_effect_ids[pending_at_entry:]
+                await existing.run_rollback_effects(rollback_at)
                 raise
             return
 
@@ -428,6 +430,7 @@ def blocking():
             committed = True
         except BaseException:
             conn.rollback()
+            await tx.run_rollback_effects()
             raise
         finally:
             tx.closed = True

src/msg/storage/sqlite.py

@@ -200,6 +200,25 @@ def __init__(self, connection, *, write: bool):
         self._connection, self.write = connection, write
         self.owner_task = asyncio.current_task()
         self.closed = False
+        self.rollback_effects = []
+
+    def on_rollback(self, effect):
+        """Register an async compensating effect for work outside the database."""
+        self.check(write=True)
+        self.rollback_effects.append(effect)
+
+    async def run_rollback_effects(self, start=0):
+        effects = self.rollback_effects[start:]
+        self.rollback_effects = self.rollback_effects[:start]
+        first_error = None
+        for effect in reversed(effects):
+            try:
+                await effect()
+            except BaseException as exc:  # noqa: BLE001 - cancellation must not skip cleanup
+                if first_error is None:
+                    first_error = exc
+        if first_error is not None:
+            raise first_error
 
     def check(self, write=False):
         require(not self.closed, "transaction_closed")
@@ -567,12 +586,14 @@ async def transaction(self, *, write):
             existing.check(write)
             name='nested_'+uuid.uuid4().hex
             existing.execute('SAVEPOINT '+name)
+            rollback_at=len(existing.rollback_effects)
             try:
                 yield existing
                 existing.execute('RELEASE SAVEPOINT '+name)
             except BaseException:
                 existing.execute('ROLLBACK TO SAVEPOINT '+name)
                 existing.execute('RELEASE SAVEPOINT '+name)
+                await existing.run_rollback_effects(rollback_at)
                 raise
             return
         conn=self._connect()
@@ -600,6 +621,7 @@ async def transaction(self, *, write):
         except BaseException:
             if conn.in_transaction:
                 conn.execute("ROLLBACK")
+            await tx.run_rollback_effects()
             raise
         finally:
             tx.closed=True

tests/test_service.py

@@ -20,11 +20,11 @@
 
 
 async def call(app, op, args, *, key=None, subject=None, certs=(), expected=(), rid=None, token=None,
-               contract_version=1):
+               contract_version=1, return_fields=()):
     packet = request_for(op, args, app.settings.service_url,
         signer=key, subject=subject, certificates=certs, expected=expected,
         request_id=rid, token=token, expires_at=NOW+timedelta(seconds=120),
-        contract_version=contract_version)
+        contract_version=contract_version, return_fields=return_fields)
     return await app.executor.execute(packet)
 
 

tests/test_store_foundation.py

@@ -115,6 +115,28 @@ async def test_deposit_is_immutable_and_only_seller_can_read_payload(installed):
     assert old.data['listing']['terms_digest'] != newer.data['listing']['terms_digest']
 
 
+@pytest.mark.asyncio
+async def test_failed_deposit_projection_removes_external_pins(installed):
+    app,_=installed
+    key,seller,_=await register(app,'deposit-rollback')
+    listing=await call(app,'store.listing_create',listing_args(),key=key,subject=seller)
+    file=await call(app,'content.file_put',{'parent':'/@deposit-rollback/files',
+        'name':'private.bin','data':b64(b'x')},key=key,subject=seller)
+    payload={'id':file.resources[0].id,'revision':file.resources[0].revision}
+    pin_dir=app.contents.path/'pins'
+    before={path for path in pin_dir.glob('*/*') if path.is_file()}
+    for _ in range(2):
+        failed=await call(app,'store.package_deposit',{
+            'listing_id':listing.resources[0].id,
+            'listing_revision':listing.resources[0].revision,
+            'manifest':{},'payload_refs':[payload]},key=key,subject=seller,
+            rid='failed-deposit',return_fields=('field_that_does_not_exist',))
+        assert failed.error.code=='unknown_projection_field'
+        assert {path for path in pin_dir.glob('*/*') if path.is_file()}==before
+    async with app.metadata.transaction(write=False) as tx:
+        assert tx.one('SELECT COUNT(*) FROM store_packages')[0]==0
+
+
 @pytest.mark.asyncio
 async def test_listing_cas_payload_ownership_and_no_secret_leak(installed):
     app,_=installed