Backups: a revision chain belongs to a PC, not just a machine

CHANGELOG.md

@@ -21,6 +21,17 @@ ADR-007 and ADR-002.
   number now identifies the machine only. A new PC takes its hostname as its
   asset number, which is what the existing data already does, and an existing
   PC's asset number is never overwritten.
+- Configuration backups piled up duplicate revisions on any machine number
+  reported by more than one PC. A posted config was compared against the latest
+  revision for the machine, so where several PCs share a number - the part
+  markers do, and their configs differ by COM port - each PC's post differed
+  from whichever PC posted last, nothing deduped, and the table grew by a row
+  per PC per collection cycle. A backup chain is now per source PC, so an
+  unchanged config is a no-op again and each PC keeps its own history against
+  the machine it belongs to. Retention prunes each chain separately, so a busy
+  PC's revisions can no longer push out a quiet PC's only backup, and the
+  revision diff compares against the same PC's previous revision rather than
+  another PC's.
 - A backup export containing a value with an empty right-hand side (`Name=`)
   failed to parse, and with it the whole file, so that machine could never be
   backed up. The form is not strictly legal but occurs in real exports. It is

plugins/backups/api/routes.py

@@ -27,6 +27,7 @@ from shopdb.api import (
 from ..models import BackupRevision
 from ..models.backup import _utciso
 from ..services.registry import REGISTRY, getkind
+from ..services.retention import samesource
 
 backups_bp = Blueprint('backups', __name__)
 
@@ -278,9 +279,13 @@ def diff_revision(backuprevisionid):
     if againstid:
         other = db.session.get(BackupRevision, againstid)
     else:
+        # The previous revision FROM THE SAME PC. Several PCs can share a
+        # machine number, and diffing across them reported one device's COM port
+        # as a change on another, which is not a change at all.
         other = (db.session.query(BackupRevision)
                  .filter(BackupRevision.assetid == revision.assetid,
                          BackupRevision.backupkind == revision.backupkind,
+                         samesource(revision.sourcehostname),
                          BackupRevision.backuprevisionid < revision.backuprevisionid)
                  .order_by(BackupRevision.backuprevisionid.desc())
                  .first())

plugins/backups/plugin.py

@@ -25,7 +25,7 @@ from .api import backups_bp
 from .models import BackupRevision
 from .services.registry import (REGISTRY, getkind, canonicalhash,
                                DEFAULTSHAREROOT)
-from .services.retention import prune
+from .services.retention import prune, samesource as _samesource
 
 logger = logging.getLogger(__name__)
 
@@ -266,15 +266,26 @@ class BackupsPlugin(BasePlugin):
                     "contenthash and sharepath are required for kind '{}'".format(
                         kindkey))
 
-        # with_for_update serializes concurrent posts for the same asset+kind.
+        # Dedup is against the LATEST row of THIS SOURCE's chain, not the
-        # The whole fleet collects on the same GE-Enforce cycle, so a retry or
+        # asset's. Several PCs legitimately share one machine number here - the
-        # two PCs claiming one machine number can otherwise both pass the hash
+        # part markers on 0613, 0615 and WJPRT do, and their configs genuinely
-        # check and insert duplicate identical revisions. No unique constraint
+        # differ, typically by COM port. Keyed on
-        # can cover this: dedup is against the LATEST row only, because a config
+        # asset alone, each marker's post differed from whichever marker posted
+        # last, so nothing ever deduped and the chain grew by one revision per
+        # PC per collection cycle. Keyed on the source, an unchanged config is a
+        # no-op again and each marker keeps its own history against the machine.
+        #
+        # Dedup is against the latest row only, not any row, because a config
         # reverting to an earlier state is a legitimate new revision.
+        #
+        # with_for_update serializes concurrent posts for the same chain: the
+        # fleet collects on one GE-Enforce cycle, so a retry could otherwise
+        # pass the hash check twice and insert the same revision.
+        sourcehostname = payload.get('sourcehostname')
         latest = (db.session.query(BackupRevision)
                   .filter(BackupRevision.assetid == assetid,
-                          BackupRevision.backupkind == kindkey)
+                          BackupRevision.backupkind == kindkey,
+                          _samesource(sourcehostname))
                   .order_by(BackupRevision.backuprevisionid.desc())
                   .with_for_update()
                   .first())

plugins/backups/services/retention.py

@@ -2,18 +2,23 @@
 
 Dedup already keeps growth low - a revision appears only when a setting really
 changed - so retention exists for the pathological case, not the normal one: a
-value that flaps (or two PCs alternately claiming one machine number) would
+value that flaps would otherwise append a revision every collection cycle
-otherwise append a revision every collection cycle forever.
+forever.
 
 Two independent limits, both off by default at 0:
 
-  retentioncount  keep at most N revisions per asset per kind
+  retentioncount  keep at most N revisions per chain
   retentiondays   additionally drop anything older than N days
 
-The NEWEST and the OLDEST revision are never pruned. The newest is the one a
+A CHAIN is one asset, one kind, one source hostname. Several PCs can share a
-tech restores from; the oldest is the earliest known-good baseline, which is
+machine number, and each keeps its own revision history against that machine;
-usually the most valuable row in the table and the one a naive "keep last N"
+pruning per asset would let a busy marker's revisions push a quiet marker's
-would delete first.
+only backup out of the table.
+
+The NEWEST and the OLDEST revision of each chain are never pruned. The newest
+is the one a tech restores from; the oldest is the earliest known-good
+baseline, which is usually the most valuable row in the table and the one a
+naive "keep last N" would delete first.
 """
 
 from datetime import datetime, timedelta
@@ -23,11 +28,45 @@ from shopdb.api import db
 from ..models import BackupRevision
 
 
+def samesource(sourcehostname):
+    """Filter matching one source's chain, treating NULL as its own source.
+
+    `sourcehostname == None` never matches in SQL, so a NULL-sourced row - every
+    row written before the column was populated, and any hand-loaded one - needs
+    IS NULL. Without this those rows form no chain at all and re-post forever.
+    """
+    if sourcehostname is None or sourcehostname == '':
+        return BackupRevision.sourcehostname.is_(None)
+    return BackupRevision.sourcehostname == sourcehostname
+
+
+def sourcesfor(assetid, backupkind):
+    """Every distinct source hostname with a revision for this asset+kind."""
+    rows = (db.session.query(BackupRevision.sourcehostname)
+            .filter(BackupRevision.assetid == assetid,
+                    BackupRevision.backupkind == backupkind)
+            .distinct()
+            .all())
+    return [row[0] for row in rows]
+
+
 def prune(assetid, backupkind, retentioncount=0, retentiondays=0):
     """Delete surplus revisions for one asset+kind. Returns the number removed.
 
-    Caller commits. Returns 0 when both limits are disabled.
+    Prunes each source's chain independently. Caller commits. Returns 0 when
+    both limits are disabled.
     """
+    if int(retentioncount or 0) <= 0 and int(retentiondays or 0) <= 0:
+        return 0
+    return sum(prunechain(assetid, backupkind, source,
+                          retentioncount=retentioncount,
+                          retentiondays=retentiondays)
+               for source in sourcesfor(assetid, backupkind))
+
+
+def prunechain(assetid, backupkind, sourcehostname, retentioncount=0,
+               retentiondays=0):
+    """Delete surplus revisions for one asset+kind+source chain."""
     retentioncount = int(retentioncount or 0)
     retentiondays = int(retentiondays or 0)
     if retentioncount <= 0 and retentiondays <= 0:
@@ -35,7 +74,8 @@ def prune(assetid, backupkind, retentioncount=0, retentiondays=0):
 
     revisions = (db.session.query(BackupRevision)
                  .filter(BackupRevision.assetid == assetid,
-                         BackupRevision.backupkind == backupkind)
+                         BackupRevision.backupkind == backupkind,
+                         samesource(sourcehostname))
                  .order_by(BackupRevision.backuprevisionid.desc())
                  .all())
     if len(revisions) <= 2:

scripts/collapse_duplicate_backup_revisions.py

@@ -0,0 +1,203 @@
+"""Collapse the duplicate backup revisions left by the shared-machine-number bug.
+
+Dedup used to compare a posted config against the latest revision for the
+ASSET. That is correct only when a machine number means one PC. Several PCs
+legitimately share one here - the part markers on 0613, 0615 and WJPRT, whose
+configs differ by COM port - so each marker's post differed from whichever
+marker had posted last, nothing ever deduped, and the table grew by one row per
+PC per collection cycle. Dedup is now per source hostname; this cleans up what
+the old rule wrote.
+
+A machine number that is NOT meant to be shared but shows two sources here is a
+different fault - two PCs carrying the same machine number - and this script
+will report it as two chains. Deal with that at the PC, not in the database.
+
+WHAT IT REMOVES, and nothing else: within one chain (asset + kind + source
+hostname), a revision whose content hash equals the revision immediately before
+it from the same source. Those rows record no change and are exactly the rows
+the fixed code would not have created.
+
+WHAT IT KEEPS: every genuine change, in order; the newest revision of every
+chain; the oldest revision of every chain; and every chain of every source, so
+a quiet PC's only backup is never touched. A config that reverts to an earlier
+state keeps both rows, because that is a real change and not a duplicate.
+
+Share-backed kinds (partmarker) are metadata rows pointing at files on the
+share. Deleting a row does not touch the file.
+
+Usage (on the target instance, in the app dir):
+    venv\\Scripts\\python scripts\\collapse_duplicate_backup_revisions.py
+    venv\\Scripts\\python scripts\\collapse_duplicate_backup_revisions.py --commit
+    ... --assetnumber 0615        # limit to one machine
+    ... --kind ntlars             # limit to one kind
+
+Dry run by default: it prints what it would delete and changes nothing. Take a
+database backup before --commit anyway; this deletes rows.
+"""
+
+import argparse
+import os
+import sys
+from collections import defaultdict
+
+# Run from anywhere: put the repo root (parent of scripts/) on the path.
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from shopdb import create_app
+from shopdb.api import db
+from shopdb.core.models import Asset
+from plugins.backups.models import BackupRevision
+
+
+def _chains(assetnumber=None, kind=None):
+    """Every revision grouped into (assetid, kind, sourcehostname) chains."""
+    query = db.session.query(BackupRevision)
+    if assetnumber:
+        asset = db.session.query(Asset).filter(
+            Asset.assetnumber == assetnumber).first()
+        if asset is None:
+            raise SystemExit('no asset with assetnumber {!r}'.format(assetnumber))
+        query = query.filter(BackupRevision.assetid == asset.assetid)
+    if kind:
+        query = query.filter(BackupRevision.backupkind == kind)
+
+    chains = defaultdict(list)
+    for revision in query.order_by(BackupRevision.backuprevisionid.asc()).all():
+        key = (revision.assetid, revision.backupkind, revision.sourcehostname)
+        chains[key].append(revision)
+    return chains
+
+
+def _redundant(revisions):
+    """Revisions in one chain that repeat the hash of the one before them.
+
+    The chain is in id order, so this walks it forwards and keeps the first
+    appearance of each run. A later run of the same hash after a genuine change
+    is kept: that is a revert, which is a real event.
+    """
+    doomed = []
+    previoushash = None
+    for revision in revisions:
+        if previoushash is not None and revision.contenthash == previoushash:
+            doomed.append(revision)
+        else:
+            previoushash = revision.contenthash
+    # Never drop the newest row of a chain, even if it repeats: it is the row
+    # the asset panel and a restore both read.
+    newestid = revisions[-1].backuprevisionid
+    return [r for r in doomed if r.backuprevisionid != newestid]
+
+
+def _report(chains):
+    """Say what grew each chain, so a real cause can be told from the shared
+    machine numbers.
+
+    Three findings, and they need different fixes:
+
+      several sources on one machine number - either a legitimately shared
+      number (0613, 0615, WJPRT) or, on any other number, two PCs carrying the
+      same machine number, which is a fault to fix on the PC.
+
+      one source, many revisions, few distinct hashes - the old asset-wide
+      dedup, or a value flapping between two states.
+
+      one source, many revisions, all distinct hashes - something in the
+      captured config changes on its own. The value names listed are the ones
+      that changed most often between consecutive revisions; a timestamp or a
+      session counter in the projection would show up here, and the fix is to
+      exclude it from the hash, not to delete rows.
+    """
+    from plugins.backups.api.routes import _diffprojections
+
+    bymachine = defaultdict(list)
+    for key, revisions in chains.items():
+        bymachine[key[0], key[1]].append((key[2], revisions))
+
+    for (assetid, kind), sources in sorted(bymachine.items()):
+        asset = db.session.get(Asset, assetid)
+        name = asset.assetnumber if asset else str(assetid)
+        total = sum(len(r) for _, r in sources)
+        print('{} / {}: {} revisions from {} source(s)'.format(
+            name, kind, total, len(sources)))
+
+        for source, revisions in sorted(sources, key=lambda s: s[0] or ''):
+            hashes = {r.contenthash for r in revisions}
+            print('    {:<20} {:>5} revisions, {:>4} distinct'.format(
+                source or '(no source)', len(revisions), len(hashes)))
+
+            if len(revisions) < 3 or kind != 'ntlars':
+                continue
+            churn = defaultdict(int)
+            for older, newer in zip(revisions, revisions[1:]):
+                if older.contenthash == newer.contenthash:
+                    continue
+                for change in _diffprojections(older.payload, newer.payload):
+                    churn['{}\\{}'.format(change['keypath'],
+                                          change['valuename'])] += 1
+            for valuename, count in sorted(churn.items(),
+                                           key=lambda kv: -kv[1])[:5]:
+                print('        changed {:>4}x  {}'.format(count, valuename))
+        print()
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--commit', action='store_true',
+                        help='apply the deletions (default is a dry run)')
+    parser.add_argument('--report', action='store_true',
+                        help='explain what grew each chain and delete nothing')
+    parser.add_argument('--assetnumber', help='limit to one machine number')
+    parser.add_argument('--kind', help='limit to one backup kind')
+    args = parser.parse_args()
+
+    app = create_app(os.environ.get('FLASK_ENV', 'production'))
+    with app.app_context():
+        chains = _chains(args.assetnumber, args.kind)
+        if not chains:
+            print('No revisions matched.')
+            return
+
+        if args.report:
+            _report(chains)
+            return
+
+        assetnames = {}
+        total = 0
+        machines = defaultdict(int)
+
+        for (assetid, kind, source), revisions in sorted(chains.items(),
+                                                         key=lambda kv: kv[0]):
+            doomed = _redundant(revisions)
+            if not doomed:
+                continue
+            if assetid not in assetnames:
+                asset = db.session.get(Asset, assetid)
+                assetnames[assetid] = (asset.assetnumber if asset
+                                       else str(assetid))
+            name = assetnames[assetid]
+            machines[name] += len(doomed)
+            total += len(doomed)
+            print('{:>10}  {:<12} {:<20} {} of {} redundant'.format(
+                name, kind, source or '(no source)', len(doomed),
+                len(revisions)))
+            for revision in doomed:
+                db.session.delete(revision)
+
+        print()
+        for name in sorted(machines):
+            print('{:>10}: {} rows'.format(name, machines[name]))
+        print('{} chains, {} redundant revisions'.format(len(chains), total))
+
+        if not total:
+            print('Nothing to do.')
+            return
+        if args.commit:
+            db.session.commit()
+            print('DELETED {} revisions.'.format(total))
+        else:
+            db.session.rollback()
+            print('Dry run - nothing changed. Re-run with --commit to apply.')
+
+
+if __name__ == '__main__':
+    main()

tests/test_plugins/test_backups.py

@@ -757,3 +757,93 @@ def test_a_marker_export_shows_the_mark_tab():
     assert any('MARK' in label for label in labels)
     mark = next(s for s in card['sections'] if 'MARK' in s['label'])
     assert any(f['label'] == 'Port Id' and f['value'] == 'COM4' for f in mark['fields'])
+
+
+# =============================================================================
+# Several PCs sharing one machine number
+# =============================================================================
+
+def _markerreg(comport):
+    """A marker config that differs from its sibling only by COM port.
+
+    Which is what the real ones do: the part markers sharing 0613, 0615 and
+    WJPRT are separate devices on separate ports, filed under one machine
+    number.
+    """
+    return CONFIGUREDREG.replace('"MachineNo"="3204"',
+                                 '"MachineNo"="3204"\r\n'
+                                 '"Port Id"="{}"'.format(comport))
+
+
+def test_two_pcs_on_one_machine_number_do_not_thrash(bk_app, bk_plugin):
+    """The bug that filled the table: dedup was against the LATEST revision for
+    the asset, so with two PCs on one machine number each post differed from
+    whichever PC posted last. Nothing ever deduped and the chain grew by a row
+    per PC per collection cycle, forever."""
+    from plugins.backups.models import BackupRevision
+    with bk_app.app_context():
+        for _ in range(5):
+            bk_plugin.apply_collector_payload(
+                _payload(reg=_markerreg('COM3'), sourcehostname='MARKERA'))
+            bk_plugin.apply_collector_payload(
+                _payload(reg=_markerreg('COM4'), sourcehostname='MARKERB'))
+
+        # One revision each, not ten.
+        assert _db.session.query(BackupRevision).count() == 2
+
+
+def test_each_pc_keeps_its_own_history_on_a_shared_machine(bk_app, bk_plugin):
+    """Both markers' configs matter, so both chains are kept in full."""
+    from plugins.backups.models import BackupRevision
+    with bk_app.app_context():
+        bk_plugin.apply_collector_payload(
+            _payload(reg=_markerreg('COM3'), sourcehostname='MARKERA'))
+        bk_plugin.apply_collector_payload(
+            _payload(reg=_markerreg('COM4'), sourcehostname='MARKERB'))
+        # MARKERA is re-cabled. That is a real change on A, and none on B.
+        bk_plugin.apply_collector_payload(
+            _payload(reg=_markerreg('COM5'), sourcehostname='MARKERA'))
+        result = bk_plugin.apply_collector_payload(
+            _payload(reg=_markerreg('COM4'), sourcehostname='MARKERB'))
+
+        assert result['action'] == 'noop'
+        rows = _db.session.query(BackupRevision).all()
+        bysource = {}
+        for row in rows:
+            bysource.setdefault(row.sourcehostname, []).append(row)
+        assert len(bysource['MARKERA']) == 2
+        assert len(bysource['MARKERB']) == 1
+
+
+def test_a_null_source_still_forms_a_chain(bk_app, bk_plugin):
+    """Rows written before sourcehostname was populated, and hand-loaded ones,
+    carry NULL. `column == None` never matches in SQL, so without an IS NULL
+    those rows would dedupe against nothing and re-post every cycle."""
+    from plugins.backups.models import BackupRevision
+    with bk_app.app_context():
+        first = bk_plugin.apply_collector_payload(_payload(sourcehostname=None))
+        second = bk_plugin.apply_collector_payload(_payload(sourcehostname=None))
+        assert second['action'] == 'noop'
+        assert second['backuprevisionid'] == first['backuprevisionid']
+        assert _db.session.query(BackupRevision).count() == 1
+
+
+def test_retention_prunes_each_pc_separately(bk_app, bk_plugin):
+    """Pruning per asset let a busy marker's revisions evict a quiet marker's
+    only backup, which is the one row that mattered on that PC."""
+    from plugins.backups.models import BackupRevision
+    from plugins.backups.services.retention import prune
+    with bk_app.app_context():
+        bk_plugin.apply_collector_payload(
+            _payload(reg=_markerreg('COM9'), sourcehostname='QUIET'))
+        for port in ('COM3', 'COM4', 'COM5', 'COM6', 'COM7'):
+            bk_plugin.apply_collector_payload(
+                _payload(reg=_markerreg(port), sourcehostname='BUSY'))
+
+        asset = _db.session.query(BackupRevision).first().assetid
+        prune(asset, 'ntlars', retentioncount=2)
+        _db.session.commit()
+
+        rows = _db.session.query(BackupRevision).all()
+        sources = [row.sourcehostname for row in rows]
+        assert 'QUIET' in sources, 'the quiet PC lost its only backup'