diff --git a/morango/sync/context.py b/morango/sync/context.py
index 3c4d72d3..4319cbd5 100644
--- a/morango/sync/context.py
+++ b/morango/sync/context.py
@@ -4,6 +4,7 @@
 from morango.models.certificates import Filter
 from morango.models.core import SyncSession
 from morango.models.core import TransferSession
+from morango.sync.db import SyncDatabase
 from morango.utils import CAPABILITIES
 from morango.utils import parse_capabilities_from_server_request
 
@@ -20,6 +21,7 @@ class SessionContext(object):
         "is_push",
         "capabilities",
         "error",
+        "db",
     )
     max_backoff_interval = 5
 
@@ -30,6 +32,7 @@ def __init__(
         sync_filter=None,
         is_push=None,
         capabilities=None,
+        db=None,
     ):
         """
         :param sync_session: The sync session instance
@@ -42,6 +45,8 @@ def __init__(
         :type is_push: bool
         :param capabilities: Capabilities set that is combined (intersected) with our own capabilities
         :type capabilities: set|None
+        :param db: Database handle for local sync operations, defaulting to routed access
+        :type db: SyncDatabase|None
         """
         self.sync_session = sync_session
         self.transfer_session = transfer_session
@@ -49,6 +54,7 @@ def __init__(
         self.is_push = is_push
         self.capabilities = set(capabilities or []) & CAPABILITIES
         self.error = None
+        self.db = db if db is not None else SyncDatabase.default()
 
         if self.transfer_session:
             self.sync_session = transfer_session.sync_session or self.sync_session
@@ -179,6 +185,7 @@ def __getstate__(self):
 
     def __setstate__(self, state):
         """Re-apply dict state after serialization"""
+        self.db = SyncDatabase.default()
         sync_session_id = state.get("sync_session_id", None)
         if sync_session_id is not None:
             self.sync_session = SyncSession.objects.get(pk=sync_session_id)
@@ -498,6 +505,7 @@ def __getstate__(self):
 
     def __setstate__(self, state):
         """Re-apply dict state after serialization"""
+        self.db = SyncDatabase.default()
         self.children = state.get("children", [])
         self._counter = state.get("counter", 0)
         self._stage = state.get("stage", None)
diff --git a/morango/sync/operations.py b/morango/sync/operations.py
index 65059c1b..86e78b4e 100644
--- a/morango/sync/operations.py
+++ b/morango/sync/operations.py
@@ -25,10 +25,10 @@
 from morango.models.certificates import Filter
 from morango.models.core import Buffer
 from morango.models.core import DatabaseMaxCounter
-from morango.models.core import InstanceIDModel
 from morango.models.core import RecordMaxCounter
 from morango.models.core import RecordMaxCounterBuffer
 from morango.models.core import Store
+from morango.models.core import SyncSession
 from morango.models.core import TransferSession
 from morango.models.core import UUIDField
 from morango.models.fsic_utils import calculate_directional_fsic_diff
@@ -332,13 +332,13 @@ def _queue_into_buffer_v2(transfersession, chunk_size=200):
             )
 
 
-def _update_legacy_self_ref_order_for_model(queryset):
+def _update_legacy_self_ref_order_for_model(queryset, db):
     # root nodes set the _self_ref_order to 0
     queryset.filter(_self_ref_fk="").exclude(_self_ref_order=0).update(_self_ref_order=0)
     # reset the _self_ref_order to None for all records that have a parent
     queryset.exclude(_self_ref_fk="").exclude(_self_ref_order=None).update(_self_ref_order=None)
 
-    parent = Store.objects.filter(
+    parent = db.manager(Store).filter(
         id=Cast(OuterRef("_self_ref_fk"), UUIDField()),
         _self_ref_order__isnull=False,
     )
@@ -349,9 +349,9 @@ def _update_legacy_self_ref_order_for_model(queryset):
         pass
 
 
-def _update_legacy_self_ref_order(transfer_session):
-    profile = transfer_session.sync_session.profile
-    transferred_store_records = Store.objects.filter(
+def _update_legacy_self_ref_order(transfer_session, db):
+    profile = db.manager(SyncSession).get(pk=transfer_session.sync_session_id).profile
+    transferred_store_records = db.manager(Store).filter(
         last_transfer_session_id=transfer_session.id,
         profile=profile,
     )
@@ -359,35 +359,42 @@ def _update_legacy_self_ref_order(transfer_session):
     for Model in syncable_models.get_models(profile):
         queryset = transferred_store_records.filter(model_name=Model.morango_model_name)
         if self_referential_fk(Model):
-            _update_legacy_self_ref_order_for_model(queryset)
+            _update_legacy_self_ref_order_for_model(queryset, db)
         else:
             queryset.exclude(_self_ref_order=None).update(_self_ref_order=None)
 
 
-def _dequeue_into_store(transfer_session, fsic, v2_format=False, self_ref_order=True):
+def _dequeue_into_store(transfer_session, fsic, v2_format, self_ref_order, db):
     """
     Takes data from the buffers and merges into the store and record max counters.
 
+    :param transfer_session: Transfer session whose buffered records will be merged
+    :param fsic: JSON-encoded filter-specific instance counters from the sender
+    :param v2_format: Whether the counters use the v2 format
+    :param self_ref_order: Whether the sender supplied self-referential ordering
+    :param db: Database handle for the dequeue transaction, SQL, and instance counter
+    :type db: morango.sync.db.SyncDatabase
+
     ALGORITHM: Incrementally insert and delete on a case by case basis to ensure subsequent cases
     are not affected by previous cases.
     """
 
-    with begin_transaction(Filter(transfer_session.filter)):
-        with connection.cursor() as cursor:
-            DBBackend._dequeuing_delete_rmcb_records(cursor, transfer_session.id)
-            DBBackend._dequeuing_delete_buffered_records(cursor, transfer_session.id)
-            current_id = InstanceIDModel.get_current_instance_and_increment_counter()
-            DBBackend._dequeuing_merge_conflict_buffer(cursor, current_id, transfer_session.id)
-            DBBackend._dequeuing_merge_conflict_rmcb(cursor, transfer_session.id)
-            DBBackend._dequeuing_update_rmcs_last_saved_by(cursor, current_id, transfer_session.id)
-            DBBackend._dequeuing_delete_mc_rmcb(cursor, transfer_session.id)
-            DBBackend._dequeuing_delete_mc_buffer(cursor, transfer_session.id)
-            DBBackend._dequeuing_insert_remaining_buffer(cursor, transfer_session.id)
-            DBBackend._dequeuing_insert_remaining_rmcb(cursor, transfer_session.id)
+    with db.begin_transaction(Filter(transfer_session.filter)):
+        with db.cursor() as cursor:
+            db.backend._dequeuing_delete_rmcb_records(cursor, transfer_session.id)
+            db.backend._dequeuing_delete_buffered_records(cursor, transfer_session.id)
+            current_id = db.current_instance_and_increment_counter()
+            db.backend._dequeuing_merge_conflict_buffer(cursor, current_id, transfer_session.id)
+            db.backend._dequeuing_merge_conflict_rmcb(cursor, transfer_session.id)
+            db.backend._dequeuing_update_rmcs_last_saved_by(cursor, current_id, transfer_session.id)
+            db.backend._dequeuing_delete_mc_rmcb(cursor, transfer_session.id)
+            db.backend._dequeuing_delete_mc_buffer(cursor, transfer_session.id)
+            db.backend._dequeuing_insert_remaining_buffer(cursor, transfer_session.id)
+            db.backend._dequeuing_insert_remaining_rmcb(cursor, transfer_session.id)
             if not self_ref_order:
-                _update_legacy_self_ref_order(transfer_session)
-            DBBackend._dequeuing_delete_remaining_rmcb(cursor, transfer_session.id)
-            DBBackend._dequeuing_delete_remaining_buffer(cursor, transfer_session.id)
+                _update_legacy_self_ref_order(transfer_session, db)
+            db.backend._dequeuing_delete_remaining_rmcb(cursor, transfer_session.id)
+            db.backend._dequeuing_delete_remaining_buffer(cursor, transfer_session.id)
 
         DatabaseMaxCounter.update_fsics(
             json.loads(fsic),
@@ -712,6 +719,7 @@ def handle(self, context):
                 fsic,
                 v2_format=FSIC_V2_FORMAT in context.capabilities,
                 self_ref_order=SELF_REF_ORDER in context.capabilities,
+                db=context.db,
             )
 
         return transfer_statuses.COMPLETED
diff --git a/tests/testapp/tests/sync/test_context.py b/tests/testapp/tests/sync/test_context.py
index f9941a19..fb708170 100644
--- a/tests/testapp/tests/sync/test_context.py
+++ b/tests/testapp/tests/sync/test_context.py
@@ -15,12 +15,18 @@
 from morango.sync.context import NetworkSessionContext
 from morango.sync.context import SessionContext
 from morango.sync.controller import SessionController
+from morango.sync.db import SyncDatabase
 
 from ..helpers import create_dummy_store_data
 from ..helpers import TestSessionContext
 
 
 class SessionContextTestCase(SimpleTestCase):
+    def test_database_handle(self):
+        self.assertIsNone(TestSessionContext().db.alias)
+        db = SyncDatabase("default2")
+        self.assertIs(TestSessionContext(db=db).db, db)
+
     def test_init__nothing(self):
         context = TestSessionContext()
         context.update_state(
@@ -259,6 +265,15 @@ def test_init(self):
 
 
 class ContextPicklingTestCase(TestCase):
+    def test_local_restores_default_database_handle(self):
+        context = LocalSessionContext(db=SyncDatabase("default2"))
+        self.assertNotIn("db", context.__getstate__())
+
+        restored = pickle.loads(pickle.dumps(context))
+
+        self.assertIsInstance(restored.db, SyncDatabase)
+        self.assertIsNone(restored.db.alias)
+
     def test_basic(self):
         data = create_dummy_store_data()
         transfer_session = data["tx"]
@@ -319,7 +334,7 @@ def test_composite(self):
         )
         network = NetworkSessionContext(conn)
 
-        composite = CompositeSessionContext([local, network])
+        composite = CompositeSessionContext([local, network], db=SyncDatabase("default2"))
         for _ in range(2):
             composite.update_state(
                 stage=transfer_stages.INITIALIZING, stage_status=transfer_statuses.COMPLETED
@@ -333,8 +348,10 @@ def test_composite(self):
         self.assertEqual(composite._counter, unpickled_context._counter)
         self.assertEqual(composite.stage, unpickled_context.stage)
         self.assertEqual(composite.stage_status, unpickled_context.stage_status)
+        self.assertIsNone(unpickled_context.db.alias)
         for i, context_type in enumerate([LocalSessionContext, NetworkSessionContext]):
             self.assertIsInstance(unpickled_context.children[i], context_type)
+            self.assertIsNone(unpickled_context.children[i].db.alias)
 
 
 class CompositeSessionContextTestCase(SimpleTestCase):
diff --git a/tests/testapp/tests/sync/test_operations.py b/tests/testapp/tests/sync/test_operations.py
index 979c32c2..f7d472f2 100644
--- a/tests/testapp/tests/sync/test_operations.py
+++ b/tests/testapp/tests/sync/test_operations.py
@@ -2,7 +2,10 @@
 import uuid
 
 import mock
+import pytest
+from django.conf import settings
 from django.db import connection
+from django.db import transaction
 from django.test import override_settings
 from django.test import TestCase
 from django.utils import timezone
@@ -29,6 +32,7 @@
 from morango.sync.context import LocalSessionContext
 from morango.sync.controller import MorangoProfileController
 from morango.sync.controller import SessionController
+from morango.sync.db import SyncDatabase
 from morango.sync.operations import _dequeue_into_store
 from morango.sync.operations import _queue_into_buffer_v1
 from morango.sync.operations import _queue_into_buffer_v2
@@ -43,6 +47,7 @@
 from morango.sync.stream.deserialize import deserialize_from_store
 from morango.sync.syncsession import TransferClient
 
+from ..helpers import assert_no_default_db_queries
 from ..helpers import create_buffer_and_store_dummy_data
 from ..helpers import create_dummy_store_data
 
@@ -673,6 +678,7 @@ def setUp(self):
             sync_session=self.transfer_session.sync_session,
             is_server=True,
             capabilities=[],
+            db=SyncDatabase.default(),
         )
 
     def assert_store_records_tagged_with_last_session(self, store_ids):
@@ -707,7 +713,11 @@ def test_dequeuing_sets_last_session(self):
         store_ids = [self.data[key] for key in ["model2", "model3", "model4", "model5", "model7"]]
         self.assert_store_records_not_tagged_with_last_session(store_ids)
         _dequeue_into_store(
-            self.transfer_session, self.transfer_session.client_fsic, v2_format=False
+            self.transfer_session,
+            self.transfer_session.client_fsic,
+            v2_format=False,
+            self_ref_order=True,
+            db=SyncDatabase.default(),
         )
         # this one is a reverse fast forward, so it doesn't modify the store record and shouldn't be tagged
         self.assert_store_records_not_tagged_with_last_session([self.data["model1"]])
@@ -939,7 +949,11 @@ def test_dequeuing_delete_remaining_buffer(self):
 
     def test_dequeue_into_store(self):
         _dequeue_into_store(
-            self.transfer_session, self.transfer_session.client_fsic, v2_format=False
+            self.transfer_session,
+            self.transfer_session.client_fsic,
+            v2_format=False,
+            self_ref_order=True,
+            db=SyncDatabase.default(),
         )
         # ensure a record with different transfer session id is not affected
         self.assertTrue(Buffer.objects.filter(transfer_session_id=self.data["tfs_id"]).exists())
@@ -1023,6 +1037,7 @@ def test_dequeue_into_store__self_ref_order_fallback_for_missing_capability(self
             self.transfer_session.client_fsic,
             v2_format=False,
             self_ref_order=False,
+            db=SyncDatabase.default(),
         )
 
         self.assertEqual(Store.objects.get(id=self.data["model3"])._self_ref_order, 0)
@@ -1034,7 +1049,7 @@ def test_update_legacy_self_ref_order_nulls_non_self_ref_models(self):
             _self_ref_order=3,
         )
 
-        _update_legacy_self_ref_order(self.transfer_session)
+        _update_legacy_self_ref_order(self.transfer_session, SyncDatabase.default())
 
         store.refresh_from_db()
         self.assertIsNone(store._self_ref_order)
@@ -1050,7 +1065,7 @@ def test_update_legacy_self_ref_order_handles_deeper_self_ref_chains(self):
             _self_ref_order=None,
         )
 
-        _update_legacy_self_ref_order(self.transfer_session)
+        _update_legacy_self_ref_order(self.transfer_session, SyncDatabase.default())
 
         root.refresh_from_db()
         child.refresh_from_db()
@@ -1072,6 +1087,7 @@ def test_local_dequeue_operation(self):
     def test_local_dequeue_operation__passes_self_ref_order_capability(self, mock_dequeue):
         self.transfer_session.records_transferred = 1
         self.context.capabilities = {SELF_REF_ORDER}
+        self.context.db = SyncDatabase("default2")
         operation = ReceiverDequeueOperation()
         self.assertEqual(transfer_statuses.COMPLETED, operation.handle(self.context))
         mock_dequeue.assert_called_once_with(
@@ -1079,6 +1095,7 @@ def test_local_dequeue_operation__passes_self_ref_order_capability(self, mock_de
             self.transfer_session.client_fsic,
             v2_format=False,
             self_ref_order=True,
+            db=self.context.db,
         )
 
     @mock.patch("morango.sync.operations._dequeue_into_store")
@@ -1126,6 +1143,126 @@ def test_local_cleanup(self):
         )
 
 
+@pytest.mark.skipif("default2" not in settings.DATABASES, reason="requires a second database")
+@override_settings(DATABASE_ROUTERS=[])
+class PinnedDequeueTestCase(TestCase):
+    databases = {"default", "default2"}
+
+    def setUp(self):
+        self.db = SyncDatabase("default2")
+        self.host_instance, _ = InstanceIDModel.get_or_create_current_instance(clear_cache=True)
+        self.host_counter = self.host_instance.counter
+        database = DatabaseIDModel(id=uuid.uuid4().hex)
+        self.db.manager(DatabaseIDModel).bulk_create([database])
+        self.instance = InstanceIDModel(
+            id=uuid.uuid4().hex, database_id=database.id, counter=7, current=True
+        )
+        self.db.manager(InstanceIDModel).bulk_create([self.instance])
+        session = self.db.manager(SyncSession).create(
+            id=uuid.uuid4().hex, profile="facilitydata", last_activity_timestamp=timezone.now()
+        )
+        self.transfer_session = self.db.manager(TransferSession).create(
+            id=uuid.uuid4().hex,
+            sync_session_id=session.id,
+            last_activity_timestamp=timezone.now(),
+            filter="partition",
+            push=True,
+        )
+        self.root_id = uuid.uuid4().hex
+        self.child_id = uuid.uuid4().hex
+        self.conflict_id = uuid.uuid4().hex
+        self.sender_id = uuid.uuid4().hex
+        for record_id, parent_id in (
+            (self.root_id, ""),
+            (self.child_id, self.root_id),
+            (self.conflict_id, ""),
+        ):
+            self.db.manager(Buffer).create(
+                transfer_session_id=self.transfer_session.id,
+                model_uuid=record_id,
+                profile="facilitydata",
+                model_name="facility",
+                partition="partition",
+                serialized="buffer",
+                last_saved_instance=self.sender_id,
+                last_saved_counter=2,
+                _self_ref_fk=parent_id,
+                _self_ref_order=0 if not parent_id else 1,
+            )
+            self.db.manager(RecordMaxCounterBuffer).create(
+                transfer_session_id=self.transfer_session.id,
+                model_uuid=record_id,
+                instance_id=self.sender_id,
+                counter=2,
+            )
+        self.db.manager(Store).create(
+            id=self.conflict_id,
+            profile="facilitydata",
+            model_name="facility",
+            partition="partition",
+            serialized="local",
+            last_saved_instance=self.instance.id,
+            last_saved_counter=7,
+        )
+        self.db.manager(RecordMaxCounter).create(
+            store_model_id=self.conflict_id, instance_id=self.instance.id, counter=7
+        )
+
+    @mock.patch.object(DatabaseMaxCounter, "update_fsics")
+    def test_pinned_dequeue(self, update_fsics):
+        # FSIC helper routing is introduced by Task 6 (#367), after this dequeue refactor.
+        for self_ref_order in (True, False):
+            with self.subTest(self_ref_order=self_ref_order):
+                with self.db.atomic():
+                    with assert_no_default_db_queries():
+                        if not self_ref_order:
+                            self.db.manager(Buffer).update(_self_ref_order=None)
+                        _dequeue_into_store(
+                            self.transfer_session, "{}", False, self_ref_order, self.db
+                        )
+                        stores = self.db.manager(Store)
+                        self.assertEqual(stores.get(pk=self.root_id).serialized, "buffer")
+                        self.assertEqual(stores.get(pk=self.root_id)._self_ref_order, 0)
+                        self.assertEqual(stores.get(pk=self.child_id)._self_ref_order, 1)
+                        conflict = stores.get(pk=self.conflict_id)
+                        self.assertEqual(conflict.last_saved_instance, self.instance.id)
+                        self.assertEqual(conflict.last_saved_counter, 8)
+                        self.assertEqual(conflict.serialized, "local")
+                        self.assertEqual(conflict.conflicting_serialized_data, "buffer\n")
+                        self.assertEqual(
+                            self.db.manager(InstanceIDModel).get(pk=self.instance.pk).counter, 8
+                        )
+                        self.assertEqual(
+                            self.db.manager(RecordMaxCounter)
+                            .filter(store_model_id__in=[self.root_id, self.child_id])
+                            .count(),
+                            2,
+                        )
+                        self.assertFalse(self.db.manager(Buffer).exists())
+                        self.assertFalse(self.db.manager(RecordMaxCounterBuffer).exists())
+                    # Roll back the subtest's writes so each capability path starts with buffers.
+                    transaction.set_rollback(True, using=self.db.alias)
+
+        update_fsics.assert_has_calls([mock.call({}, Filter("partition"), v2_format=False)] * 2)
+        self.host_instance.refresh_from_db()
+        self.assertEqual(self.host_instance.counter, self.host_counter)
+        self.assertFalse(Store.objects.filter(pk__in=[self.root_id, self.child_id]).exists())
+
+    @mock.patch.object(DatabaseMaxCounter, "update_fsics")
+    def test_pinned_dequeue_rolls_back_on_failure(self, update_fsics):
+        with mock.patch.object(
+            self.db.backend, "_dequeuing_insert_remaining_rmcb", side_effect=RuntimeError("stop")
+        ):
+            with self.assertRaisesRegex(RuntimeError, "stop"), assert_no_default_db_queries():
+                _dequeue_into_store(self.transfer_session, "{}", False, True, self.db)
+
+        self.assertEqual(self.db.manager(Buffer).count(), 3)
+        self.assertFalse(self.db.manager(Store).filter(pk=self.root_id).exists())
+        self.assertEqual(self.db.manager(Store).get(pk=self.conflict_id).last_saved_counter, 7)
+        self.assertEqual(self.db.manager(InstanceIDModel).get(pk=self.instance.pk).counter, 7)
+        update_fsics.assert_not_called()
+
+
 class DeserializationTestCases(TestCase):
     def setUp(self):
         self.profile = "facilitydata"