PPPoE active-session collector + read API

[ehauer]

What

Adds a PPPoE active-session collector so MyMate can answer who is online, on which concentrator, since when — the customer data plane — without anyone touching a router.

A scheduled command sweeps every RouterOS PPPoE concentrator's /ppp/active into a new pppoe_sessions table on a 5-minute staggered cadence, and a read endpoint serves it.

How

• SweepPppoeSessions (Action) — connects via the existing RouterOsConnection client + shared-credential resolution (RouterOsTarget::fromDevice, the device's credential_id), reads /ppp/active/print, parses name→username, address, caller-id, uptime (RouterOsDuration), and wholesale-replaces that device's rows in one transaction. Any per-device failure logs and skips — never fails the batch, never leaves partial rows.
• PppoeSweepDispatcher + SweepPppoeSessionsBatchJob — shards the fleet by crc32(device_id) % 96 (same key as the poll dispatcher) onto an isolated pppoe Horizon queue (supervisor-pppoe), dispatched with queue-side delay staggered across mymate.pppoe.stagger_seconds (240s) so ~2,300 concentrators sweep as a trickle, never a thundering herd — and can only ever delay themselves, never polling/ping/discovery.
• mymate:pppoe:sweep command with --limit N / --device ID / --now / --dry-run for canary runs; scheduled everyFiveMinutes() in routes/console.php.
• GET /api/pppoe-sessions — auth:sanctum + RestrictWritesToAdmins, cursor-paginated, ?since= (ISO8601 or epoch-ms) / ?device_id= / ?username= / ?q= filters, device_name + site_id join. Matches the site-links / rf-link-state / outages envelope.

V1 is RouterOS-only; SNMP-polled concentrators are excluded (RouterOS publishes no active-PPPoE table over SNMP).

Config

All env()-backed under mymate.pppoe.*: enabled, name_filter (%pppoe%), queue, shards (96), stagger_seconds (240), timeout (5s), job_timeout (300s), max_sessions_per_device (4000). Horizon supervisor-pppoe processes: 10 prod / 1 local.

Verified on production

Deployed and run: 9,786 sessions from 1,198 authenticating concentrators; endpoint 401s unauthed, returns joined rows with cursor pagination and working ?q= filter.

[coreyhauer]

Solid architecture — the shard/stagger design, the isolated pppoe queue, and the per-device transactional replace are the right shape, and the RouterOS-API-only scope call is correct (there is no /ppp/active over SNMP). But a deep review turned up 16 confirmed issues; the ones below need fixing before this merges. One I've already handled: the rf_link_samples partition-management fix that was running only on the box is now committed onto this branch (3dfc921) — that closes the loop on the Aug 8–10 history loss and makes the migration docblock's claim true.

Must fix — data integrity

1. persist() deletes before the empty-result guard (SweepPppoeSessions.php ~line 170). A read that succeeds at transport but yields zero usable rows (API group loses the ppp read policy; rows whose name is empty get continued away) runs the DELETE, inserts nothing, logs nothing — a healthy concentrator with hundreds of customers online reports 0 sessions, indistinguishable from truly empty. This is the exact failure the class docblock says it prevents. Move the guard above the delete, and log a warning when a previously-populated device returns empty.

2. Delete+reinsert breaks the API's own cursor pagination (PppoeSessionController.php ~line 75). Every sweep gives all rows new autoincrement ids, but the endpoint paginates by id cursor — a client walking ~13k rows across 26 pages silently loses every concentrator swept mid-walk, a different random subset each poll, with 200s all the way. Recommendation: upsert on a natural key (device_id + username/caller-id), delete only rows absent from the new read — like the sibling PollInterfaces/PollSensors do. That stabilizes ids for the cursor and removes ~7.5M dead tuples/day of churn.

3. The ?since= delta contract doesn't hold (~line 51). swept_at is stamped at shard start but rows commit up to a shard-duration later across concurrent staggered shards, so a poller that advances its cursor past a faster shard's stamp permanently misses the slower shard's rows. And disconnects emit no signal at all — a ?since= mirror shows a departed customer online forever. Either stamp at commit time and document the overlap window clients must re-read, or add a tombstone/ended_at model. (Also note: since every active row's swept_at is re-bumped each sweep, the current "delta" returns essentially the full table every 5 minutes anyway.)

Must fix — silent operational traps

4. WithoutOverlapping(...)->dontRelease() silently drops canary runs (SweepPppoeSessionsBatchJob.php ~line 52). A manual --device=X run hashes onto the same shard key the scheduled sweep holds; the job is deleted on pickup with no log, while the command prints success. Give manual runs their own lock key (or run them inline), and log when a job is dropped.

5. Producer/consumer queue-name split (config/horizon.php ~line 323). The supervisor hardcodes ['pppoe'] while jobs read config('mymate.pppoe.queue') — set MYMATE_PPPOE_QUEUE and every sweep orphans silently (~27k dead jobs/day in Redis, no Horizon failures, table just freezes). Make the supervisor read the same config key, or drop the env var.

6. No reaper for devices that leave the fleet (PppoeSweepDispatcher.php ~line 102). Fleet membership is a dispatch-time name-ILIKE; rename a concentrator (or unmonitor it, tighten the filter, disable the feature) and its last sessions are served as "currently online" forever — the endpoint has no freshness filter. Add a stale-row reaper (e.g. delete rows whose swept_at is older than N sweep intervals) and/or a default freshness filter on the endpoint.

Should fix — API robustness

7. parseSince() (~line 86): any all-digit value is treated as epoch-milliseconds, so epoch-seconds (1755000000) or ISO-basic (20260813) resolve to ~1970 and silently defeat the delta; and the createFromTimestampMs call sits outside the try/catch, so an out-of-range numeric 500s instead of 422ing. Accept both epoch scales (disambiguate by magnitude) and move the call inside the catch.

8. Unvalidated query params → 500s: ?username[]=x, ?q[]=x, ?since[]=x all reach the query builder as arrays (PDO binding mismatch / array-to-string). Same hole in SiteLinkController (?media_type=/?confidence=). RfLinkStateController already does this right with a validator returning 422 — copy that pattern.

9. Outages ?to= excludes the named day (OutageController.php ~line 59): ?to=2026-08-13 parses to midnight, so started_at <= midnight drops everything that happened on the 13th. Use endOfDay() for date-only input (or document that to is exclusive).

Worth a pass (confirmed, lower severity)

Unescaped ILIKE metacharacters in ?q= (?q=% matches everything); unbounded get() on /api/rf-link-state (~16k rows) and /api/site-links; RouterOsDuration duplicates CaptureDeviceFacts::parseRouterOsUptime; the crc32-shard loop now has a 6th copy — worth extracting; --dry-run ignores --device/--limit; check swept_at column type vs session TZ (timestamptz vs timestamp skew is deployment-dependent); the horizon env var is read twice.

---

Happy to talk through any of these. The upsert-on-natural-key change (#2) is the one that reshapes the most code, and it also simplifies #1 and #3 — I'd start there.