Fixes #6: Read Data Source directories in one query

docs/administration/configuration.md

@@ -151,10 +151,9 @@ An upload holds the existing tree plus the new file. One call can reach
 surfaces perform this work in the web process. The create and Add Script forms
 use a commit hook, while REST stages within the request.
 
-Data Source staging reads the path inventory first, then fetches content only
-under the Project's `data_path`. A repository containing several Projects does
-not load all their content for each Project. This staging, including migration
-staging, runs in an RQ worker.
+Data Source staging reads only the files under the Project's `data_path`. A
+repository containing several Projects does not load all their content for each
+Project. This staging, including migration staging, runs in an RQ worker.
 
 Verification and Script File refresh read one file at a time rather than holding
 the entire tree. Non-positive or non-integer size/count settings raise a

docs/administration/data-sources.md

@@ -35,6 +35,22 @@ Reconciliation stages the complete directory, not just changed files. Deleted
 files are absent from the next snapshot. Templates, data and other non-Python
 files are included alongside source modules.
 
+!!! warning "Reconciling during a synchronization"
+
+    NetBox does not apply a synchronization's changes all at once. A reconciliation
+    that runs while a synchronization is writing files can stage part of its
+    changes, such as changed files without the files it adds. Under **Automatic if
+    valid**, that revision is activated if it validates. The reconciliation queued
+    when that synchronization completes reads the directory again. A
+    synchronization that fails partway keeps the changes it had already written
+    until a later one completes.
+
+    Before reconciling, check the Synchronization Job's outcome in the Data
+    Source's **Jobs** tab. A failed run reads `errored`. A recurring
+    Synchronization Job sets the Data Source back to `queued` when it schedules
+    its next run, including after an error. Waiting for a Synchronization Job does
+    not stop a queued reconciliation from overlapping the next synchronization.
+
 When the directory and Script File selection are unchanged, the existing revision
 is reused. It may still be validated or activated, depending on its status and the
 Project's activation settings.
@@ -72,7 +88,8 @@ synchronization. Use it after creating a Project between synchronizations, or wh
 you need its selection applied to the current source inventory.
 
 Reconciliation does not synchronize the Data Source. To fetch newer files,
-synchronize the Data Source from its own page first.
+synchronize the Data Source from its own page first. Wait for its
+Synchronization Job to complete successfully before reconciling.
 
 This action requires the Project's `reconcile` permission, separately from
 `change`. See [Permissions](permissions.md).

docs/development/netbox-internals.md

@@ -181,6 +181,16 @@ returning. The plugin therefore creates the immediate Job itself so revision
 provenance exists before execution. A row-data argument or a documented
 post-save, pre-handler hook would remove that duplication.
 
+**Atomic Data Source synchronization.** `DataSource.sync()` commits its file
+updates, deletions and additions separately and marks the source completed
+afterwards. A reader can see part of a synchronization, and one that fails
+partway, for example on an unreadable new file, keeps the changes it already
+wrote. `ingestion.py` reads a directory's paths and contents in one query, so a
+revision holds one committed state of the inventory, but not necessarily a
+complete synchronization. Applying each synchronization in one transaction, with
+a lock on the Data Source taken before its existing files are read, would make
+every such read a complete synchronization.
+
 **A legacy export service.** Migration needs serializable module and class
 records, content checksums, Data Source references, repository paths, source
 streams and the Event Rules, permissions, Job history, schedules and recurrences

netbox_scripts/ingestion.py

@@ -215,15 +215,16 @@ def _data_source_tree(project):
     """
     Return the project's Data Source directory as a mapping of project-relative path to bytes.
 
+    The mapping is one snapshot of the inventory.
+
     Compiled artifacts are dropped, and every other unsafe path is kept so that staging records
     it rather than this function hiding it. Keys need no canonicalization of their own, because
     the store canonicalizes on the way in.
     """
-    # Two queries on purpose. The scope is decided from paths alone, so the bytes fetched are
-    # the project's directory rather than every file on the source, which for a repository
-    # holding more than this one project is the difference that matters.
-    wanted = {}
-    for pk, path in project.data_source.datafiles.values_list('pk', 'path'):
+    # Keeps other projects' bytes on the same source out of memory.
+    rows = project.data_source.datafiles.filter(path__startswith=f'{project.data_path}/')
+    tree = {}
+    for path, content in rows.values_list('path', 'data'):
         relative = data_source_relative_path(path, project.data_path)
         if relative is None:
             continue
@@ -235,11 +236,8 @@ def _data_source_tree(project):
         except UnsafePathError as error:
             if error.code == 'compiled_artifact':
                 continue
-        wanted[pk] = relative
-    if not wanted:
-        return {}
-    rows = project.data_source.datafiles.filter(pk__in=wanted).values_list('pk', 'data')
-    return {wanted[pk]: content for pk, content in rows}
+        tree[relative] = content
+    return tree
 
 
 def declare_script_file(project, path, using, *, user=None):

netbox_scripts/jobs.py

@@ -211,10 +211,10 @@ class ProjectReconciliationJob(JobRunner):
     """
     Rebuild one project's source from its Data Source directory and drive it to a verdict.
 
-    The directory is read when this runs rather than when it was enqueued, so two jobs queued by
-    two quick synchronizations are not a correctness problem: the second stages identical content
-    and resolves to the revision the first created. Queueing is not deduplicated, so the revision
-    can carry a second validation Job. The lease settles ownership, and the run that loses the
+    The directory is read when this runs rather than when it was enqueued, and unchanged content
+    resolves to the revision an earlier run created. A run that overlaps a synchronization can
+    stage part of its changes. Queueing is not deduplicated, so the revision can carry a second
+    validation Job. The lease settles ownership, and the run that loses the
     claim ends as a failed Job rather than validating the revision twice.
 
     Per project rather than per Data Source, so one project's failure leaves its siblings to

netbox_scripts/tests/test_ingestion.py

@@ -39,6 +39,7 @@
 from netbox_scripts.storage import service, store
 from netbox_scripts.storage.exceptions import StorageError
 from netbox_scripts.tests.plugin_testing import IN_MEMORY_STORAGES
+from netbox_scripts.utils import data_source_relative_path
 
 from .test_branching import routing
 
@@ -512,17 +513,41 @@ def test_the_bytes_of_a_file_outside_the_directory_are_never_fetched(self):
         # The scope is decided from paths, so a repository holding other projects does not put
         # their content in this worker's memory.
         self.populate()
-        outsider = data_file(self.source, 'unrelated/huge.py', b'z' * 4096)
+        data_file(self.source, 'unrelated/huge.py', b'z' * 4096)
 
         with CaptureQueriesContext(connection) as queries:
             tree = _data_source_tree(self.project)
 
         self.assertNotIn('huge.py', ' '.join(tree))
         fetching = [q['sql'] for q in queries.captured_queries if '"data"' in q['sql']]
         self.assertEqual(len(fetching), 1)
-        # The one query that reads bytes is restricted to the in-scope rows by primary key.
-        self.assertIn('IN (', fetching[0])
-        self.assertNotIn(str(outsider.pk), fetching[0])
+        self.assertIn('automation/netbox/%', fetching[0])
+
+    def test_paths_and_bytes_come_from_one_query(self):
+        self.populate()
+        with CaptureQueriesContext(connection) as queries:
+            _data_source_tree(self.project)
+        self.assertEqual(len([q for q in queries.captured_queries if '"core_datafile"' in q['sql']]), 1)
+
+    def test_a_synchronization_writing_during_the_read_is_not_mixed_in(self):
+        changed = data_file(self.source, 'automation/netbox/a.py', b'old')
+        removed = data_file(self.source, 'automation/netbox/obsolete.py', b'old')
+        synchronized = []
+
+        def synchronize_once(path, data_path):
+            if not synchronized:
+                synchronized.append(path)
+                DataFile.objects.filter(pk=changed.pk).update(data=b'new')
+                removed.delete()
+                data_file(self.source, 'automation/netbox/b.py', b'new')
+            return data_source_relative_path(path, data_path)
+
+        # data_source_relative_path() first runs after the reader's first query.
+        with mock.patch('netbox_scripts.ingestion.data_source_relative_path', side_effect=synchronize_once):
+            tree = _data_source_tree(self.project)
+
+        self.assertTrue(synchronized)
+        self.assertEqual(tree, {'a.py': b'old', 'obsolete.py': b'old'})
 
     def test_validation_is_enqueued_once_for_the_staged_revision(self):
         self.populate()

netbox_scripts/tests/test_reconciliation.py

@@ -6,8 +6,9 @@
 
 from django.test import TestCase, override_settings
 
-from core.choices import JobStatusChoices
+from core.choices import DataSourceStatusChoices, JobStatusChoices
 from core.exceptions import JobFailed
+from core.jobs import SyncDataSourceJob
 from core.models import DataFile, DataSource, Job
 from core.signals import post_sync
 from netbox_scripts import jobs
@@ -152,8 +153,7 @@ def test_a_file_added_after_the_enqueue_is_included(self):
         self.assertEqual(sorted(entry['path'] for entry in revision.manifest), ['audit.py', 'deploy.py'])
 
     def test_a_second_job_over_an_unchanged_directory_does_nothing(self):
-        # Two synchronizations in quick succession enqueue two jobs. The second resolves to the
-        # revision the first created, so a duplicate is a cheap no-op rather than a problem.
+        # Two synchronizations can queue two Jobs over the same content.
         self.run_job()
         revision = ScriptProjectRevision.objects.get(project=self.project)
         ScriptProjectRevision.objects.filter(pk=revision.pk).update(status=RevisionStatusChoices.VALID)
@@ -205,6 +205,26 @@ def test_a_directory_the_path_policy_refuses_fails_no_job(self):
         self.assertIn('problem(s) recorded', str(job.log_entries))
         self.validated.assert_not_called()
 
+    def test_a_recurring_source_reconciles_after_each_run(self):
+        root = Path(tempfile.mkdtemp())
+        self.addCleanup(shutil.rmtree, root, True)
+        (root / 'scripts').mkdir()
+        (root / 'scripts' / 'a.py').write_bytes(SCRIPT)
+        source = DataSource.objects.create(name='Recurring', type='local', source_url=root.as_uri())
+        project = data_source_project(source, 'recurring', 'scripts')
+        sync = SyncDataSourceJob.enqueue(instance=source, interval=60)
+        # handle() directly: immediate=True forwards job_id into the periodic re-enqueue, which rejects it.
+        SyncDataSourceJob.handle(sync)
+        sync.refresh_from_db()
+        source.refresh_from_db()
+        self.assertEqual(sync.status, JobStatusChoices.STATUS_COMPLETED)
+        self.assertEqual(source.status, DataSourceStatusChoices.QUEUED)
+
+        job = self.run_job(project.pk)
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
+        revision = ScriptProjectRevision.objects.get(project=project)
+        self.assertEqual([entry['path'] for entry in revision.manifest], ['a.py'])
+
 
 @override_settings(STORAGES=IN_MEMORY_STORAGES)
 class ReconciliationPolicyTestCase(TestCase):