Act on the wj architectural review: races, limits, pool and contract gaps
Findings F2, F4, F5, F6, F7, F8, F9, F10, F11 and F13 from wj-architectural-review.md. F1/F1a (the non-sargable LOWER pattern) and F3 (the warranty sync holding a pooled connection) are being handled separately and are untouched here. - A collector check-in that raced another for the same PC lost everything: the unique index on the application fingerprint raised, nothing caught it, and the whole check-in rolled back, discarding state, events and hardware, answering the agent with form copy meant for a person. The insert now runs in a savepoint, reconciles what the other transaction already wrote, one row per savepoint so a second collision skips a row rather than the batch, and the check-in commits with a warning. This is the shape the delete side of the same function already had. - Enrollment is deliberately unauthenticated, so it now carries a per-address rate limit and a cap on pending requests, and the cap is a setting an admin can see and change. A flood can no longer bury the approval queue or grow the table without bound. - The fleet alert poll no longer formats debug strings on every request. - BusinessUnit is on the contract surface (0.34.0), so the endpoints plugin stops reaching into a core table with raw SQL. - A test walks every asset type descriptor and proves the endpoint it names really refuses a caller without the declared permission. - The plugin settings routes now need a permission; they were telling anyone which modules a site runs and where its components live. - The connection pool is sized explicitly with a short timeout, and the arithmetic against the server's connection ceiling is written next to the worker count instead of being left to defaults nobody chose. - Fleet alerts carry both the correct lowercase keys and the old camelCase pair. Trays older than 1.7.0 match keys case-sensitively and would have gone quiet on every PC not yet upgraded; the old pair goes when they are. - The dev server proxy points at localhost again, with environment variables for the remote-box workflow, and one developer's address is out of the server certificate page.
This commit is contained in:
@@ -21,6 +21,7 @@ from pathlib import Path
|
||||
from typing import Dict, List, Optional, Type
|
||||
|
||||
from flask import Flask, Blueprint, g
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from shopdb.plugins.base import BasePlugin, PluginMeta
|
||||
from shopdb.api import db
|
||||
@@ -325,6 +326,19 @@ class EndpointsPlugin(BasePlugin):
|
||||
'description': 'True once an operator has minted a UI-managed '
|
||||
'endpoint agent key from the settings page.',
|
||||
},
|
||||
{
|
||||
'key': 'plugin.endpoints.maxpendingenrollments',
|
||||
'value': '500',
|
||||
'valuetype': 'integer',
|
||||
'category': 'plugin.endpoints',
|
||||
'description': 'How many enrollment requests may wait for an '
|
||||
'admin at once. At the cap, a request from a '
|
||||
'hostname and serial never seen before is '
|
||||
'refused; PCs already in the queue keep asking '
|
||||
'and keep being approved. 0 removes the cap. '
|
||||
'POST /enroll takes no auth, so this is what '
|
||||
'stops the pending queue growing without limit.',
|
||||
},
|
||||
]
|
||||
|
||||
def get_settings_cards(self) -> List[Dict]:
|
||||
@@ -1285,23 +1299,110 @@ class EndpointsPlugin(BasePlugin):
|
||||
f'delete count mismatch: expected {len(to_remove)}, '
|
||||
f'got {deleted_count} (concurrent update?)')
|
||||
|
||||
# Add new apps
|
||||
for fp in to_add:
|
||||
# Add new apps.
|
||||
#
|
||||
# The unique index on (assetid, appfingerprint) DETECTS a concurrent
|
||||
# insert of the same row; it does not RESOLVE one. Two check-ins for the
|
||||
# same PC can overlap (an agent retry over a slow request, two cloned
|
||||
# machines sharing a hostname, the backfill script against a live
|
||||
# fleet), and without a savepoint the IntegrityError from the flush
|
||||
# below sinks the whole check-in - state, events and hardware - which is
|
||||
# exactly the loss the delete-mismatch branch above was written to
|
||||
# avoid. So: flush inside a savepoint, and on a collision roll back only
|
||||
# the savepoint, re-read what is there now, and insert what is still
|
||||
# missing. Same shape as the health block in apply_collector_payload.
|
||||
if to_add or to_remove:
|
||||
# SAFETY: Flush to detect constraint violations before continuing.
|
||||
# Only flush if we made changes (99% of check-ins change nothing).
|
||||
try:
|
||||
with db.session.begin_nested():
|
||||
self._insert_applications(assetid, incoming, to_add)
|
||||
db.session.flush()
|
||||
except IntegrityError:
|
||||
# The context manager has already released the savepoint, so
|
||||
# the deletes and everything before them survive.
|
||||
current_app.logger.warning(
|
||||
'[collector] asset_id=%s application insert collided with a '
|
||||
'concurrent check-in (%d new fingerprint(s)); reconciling',
|
||||
assetid, len(to_add))
|
||||
return self._reconcile_applications(assetid, incoming, to_add,
|
||||
warnings)
|
||||
|
||||
return len(to_add)
|
||||
|
||||
@staticmethod
|
||||
def _insert_applications(assetid: int, incoming: Dict, fingerprints) -> list:
|
||||
"""Add one EndpointApplication per fingerprint. Returns the new rows."""
|
||||
rows = []
|
||||
for fp in fingerprints:
|
||||
app_data = incoming[fp]
|
||||
db.session.add(EndpointApplication(
|
||||
row = EndpointApplication(
|
||||
assetid=assetid,
|
||||
displayname=app_data['displayname'],
|
||||
displayversion=app_data['displayversion'],
|
||||
publisher=app_data['publisher'],
|
||||
appfingerprint=fp,
|
||||
))
|
||||
)
|
||||
rows.append(row)
|
||||
db.session.add(row)
|
||||
return rows
|
||||
|
||||
# SAFETY: Flush to detect constraint violations before continuing
|
||||
# Only flush if we made changes (skip when unchanged apps = 99% of cases)
|
||||
if to_add or to_remove:
|
||||
db.session.flush()
|
||||
def _reconcile_applications(self, assetid: int, incoming: Dict, to_add,
|
||||
warnings: List[str]) -> int:
|
||||
"""Re-insert after a unique-index collision, skipping what now exists.
|
||||
|
||||
return len(to_add)
|
||||
Called only when the bulk insert lost a race. Re-reads the asset's
|
||||
fingerprints, drops the ones another writer has since committed, and
|
||||
inserts the rest ONE ROW PER SAVEPOINT so a second collision skips that
|
||||
row instead of failing the batch again. Per-row savepoints are only
|
||||
paid on this path; the common case never reaches it.
|
||||
|
||||
Never raises: a check-in must commit its telemetry even when the
|
||||
application inventory cannot be reconciled. Whatever is missed is
|
||||
picked up by the next check-in, which reports the full list again.
|
||||
"""
|
||||
from flask import current_app
|
||||
|
||||
# The failed flush rolled the savepoint back, which restores the added
|
||||
# rows to pending. Expunge them, or the autoflush on the re-read below
|
||||
# retries the insert that just collided - this time with no savepoint
|
||||
# under it.
|
||||
for row in list(db.session.new):
|
||||
if isinstance(row, EndpointApplication) and row.assetid == assetid:
|
||||
db.session.expunge(row)
|
||||
|
||||
existing = {app.appfingerprint
|
||||
for app in (EndpointApplication.query
|
||||
.filter_by(assetid=assetid)
|
||||
.all())
|
||||
if app.appfingerprint}
|
||||
|
||||
stored = 0
|
||||
skipped = 0
|
||||
for fp in to_add:
|
||||
if fp in existing:
|
||||
skipped += 1
|
||||
continue
|
||||
try:
|
||||
with db.session.begin_nested():
|
||||
self._insert_applications(assetid, incoming, [fp])
|
||||
db.session.flush()
|
||||
stored += 1
|
||||
except IntegrityError:
|
||||
for row in list(db.session.new):
|
||||
if (isinstance(row, EndpointApplication)
|
||||
and row.appfingerprint == fp):
|
||||
db.session.expunge(row)
|
||||
skipped += 1
|
||||
|
||||
current_app.logger.warning(
|
||||
'[collector] asset_id=%s application_insert_conflict: stored=%d '
|
||||
'skipped=%d of %d new (concurrent check-in, telemetry kept)',
|
||||
assetid, stored, skipped, len(to_add))
|
||||
warnings.append(
|
||||
'application insert conflict: stored {}, skipped {} of {} new '
|
||||
'(concurrent update?)'.format(stored, skipped, len(to_add)))
|
||||
return stored
|
||||
|
||||
def apply_collector_payload(self, payload: Dict) -> Dict:
|
||||
"""Idempotently upsert one endpoint's state and append its events.
|
||||
|
||||
Reference in New Issue
Block a user