engine: close the sample-loss and duplicate cases on the write path - #1
Merged
Merged
Conversation
pvmonitor_handle now takes an on_conn callback; wire it to conn_info and the disconnect counters the way the CA monitor-end block does, so a reconnect's first sample re-arms first_after_connect and bypasses the drift filter. DBF_UCHAR lands in ScalarByte: CA promotes it to DBR_CHAR with the same 1-byte payload.
A server-side SharedPV::close() ends the handle's reactor task with Finished (pvxs sends FINISH before DESTROY_CHANNEL), and fatal/remote errors end it behind a plain Disconnected; the fast path parked on cancel_token and never re-subscribed. SubscriptionHandle has no completion future, so a drop guard captured by on_conn signals the task end and the loop retries like the custom-request path.
Rust 1.97 clippy flags the reversed sort_by closure as unnecessary_sort_by, which failed the workspace -D warnings gate.
…mple The first-after-connect waiver exists for reconnect backfill, but it also let a far-future stamp through (an IOC booted before NTP sync), and shard_handle_sample then rejected every corrected sample as out-of-order until restart. accept_ioc_timestamp is now the single gate: the 1991 floor and the future bound hold for every sample.
A failed or panicked append_event_with_meta was only an error! line, so a PV losing every sample to ENOSPC or a codec mismatch looked healthy on getPVStatus and in Prometheus. Both terminal branches of shard_handle_sample now route through record_write_error; the API DTO and the engine JSON carry the field as storageWriteErrors.
A negative secondsPastEpoch was reinterpreted as a huge u64, and Duration::new / UNIX_EPOCH + d panic on overflow. The panic unwinds inside the epics-rs reactor task that owns the subscription, so one malformed timeStamp from one IOC silently ended the monitor. Treat any unrepresentable timeStamp as "no usable timeStamp" (fall back to now), which is the documented contract of this function.
The Full case was counted but never logged and the Closed case was neither, so a PVA PV losing samples to a saturated write channel left no trace in the log to correlate with buffer_overflow_drops.
Any PbFileReader::open failure was treated as a corrupt header and the partition truncated, so a transient EMFILE/EACCES/EIO on the probe wiped every sample already in the file. Only a header that was read and failed to decode is corrupt; stat/open/read errors and a failed truncate or trim now fail the append instead.
Path::exists folds a stat error into false, so an EACCES/EIO/ENOTDIR blip on the cached writer's path discarded its dirty buffer as "file gone" although the inode was still there. try_exists keeps the buffer unless the path is positively absent.
known_dirs is a positive-only cache, so once an operator removed a PV's directory every append under that prefix failed with NotFound until restart. The open is the one place that learns the truth: on NotFound it forgets the cached parent, recreates the chain and retries once.
…PLACE REPLACE deletes the row and re-inserts only the named columns, so a repeated archivePV for an existing PV nulled last_timestamp, prec, egu, alias_for, archive_fields and policy_name. ON CONFLICT DO UPDATE touches only the columns this call owns.
rename_pv moves partitions without rewriting their PayloadInfo, so a renamed PV's files still name the old PV inside. Routing the dest partition by that header migrated the renamed PV's data under the old name, where no reader looks. The path is the identity every other ETL step already uses.
Readers decode every frame of a partition with the header's type, so a sample of another type (PV retyped at the IOC while the archiver was down, or changeTypeForPV without a partition roll) became an undecodable frame that also hid every later sample in the file. partition_header now returns the declared type and write_cached rejects the append, which surfaces as a counted write error.
The single-PV form ran an unguarded UPDATE, so it could move the last_timestamp watermark backwards while the batch form refused to. One owner of the commit rule now.
std::fs::rename replaces an existing file, so a partition already present under the new name (stale earlier rename, manual copy, a registry/disk skew the API's registry-only guard cannot see) was destroyed. Every destination is checked before the first rename so a refusal leaves the source set whole.
flush_owner_loop started its shutdown grace on the shutdown signal while the shards were still draining, so every sample the drain appended after the ~200 ms grace stayed in BufWriters with its ts_update never committed. run_sharded_write_pool now fires shards_done once every shard has returned and the owner waits for it (bounded by drain_total_budget) before the final flush. main's supervisor budget is derived from the same two timeouts so it cannot abort that final flush.
pv_name_to_key encodes ':' as '/' and left a literal '/' untouched, so A/B and A:B mapped to one partition file under two writer slots, and pv_name_from_path could not invert the key — every path-derived lookup (ETL grouping, evict, truncate) missed the raw '/' slot. Rejecting '/' makes the encoding injective; canonical_pv_key, which papered over the lossy case for the ETL paused set, goes with it.
…rating chrono's From<SystemTime> unwraps once the year leaves ±262143, so a sample with an absurd timestamp panicked inside the shard's append, the ETL move or the getDataAtTime handler. utc_datetime_checked is the one non-panicking conversion; decompose_timestamp reports the range error and the partition helpers clamp/saturate so file_path_for and the range walks stay total.
changeTypeForPV only flipped the registry row, so after a retype the current partition still carried the old header and every new sample was refused until it rolled. StoragePlugin::convert_pv_type rewrites each partition through ArchiverValue::convert_to (Java's ThruNumberConversion rules: numeric via f64, string parse/format, scalar <-> first element), to a temp sibling replaced by one rename, and skips partitions already of the new type so a re-run after a crash completes. The handler converts before the registry flips.
The header comparison ran only when write_cached (re)opened a file; a writer already cached accepted any dbr_type, so the one-type-per- partition rule held only across restarts. CachedWriter now carries the partition's type and every append is checked against it.
PvSample.dbr_type is the registry type captured once per archiving task, so the first-seen map could only ever differ for a new task started with another registry type (changeTypeForPV + resume, deletePV + re-archive) and then dropped every sample until restart. Compare the value's own type instead: that is the pair encode_sample refuses, and it is what latest_observed_dbr is meant to report.
… map The shard-lifetime last_ts map outlived the archiving task it described: after deletePV + re-archive the new task inherited the dead task's last timestamp and dropped its first samples as out-of-order. ordering_last_ts_nanos lives in the task's PvCounters, so it dies with the task; the storage writer-slot Mutex stays the real ordering owner.
If run_sharded_write_pool ends before shutdown was requested, every producer sees a closed channel and stops silently while the API keeps reporting the PVs as Active. spawn_critical turns that exit (or panic) into a shutdown request and a non-zero exit from main, and the HTTP graceful-shutdown trigger now watches the supervisor as well as SIGINT.
…_cached partition_header opens the partition before the writer does, so under fd exhaustion it was the first call to fail — outside the EMFILE evict-and-retry that only guarded the writer open, and before the budget reservation that evicts an LRU writer. Every retry then hit the same wall until fds freed elsewhere. Reserve first, and give the probe the same evict-LRU-and-retry rule.
…timeout window phase2_deadline was fixed before the shards_done wait, so a drain that spent the whole drain_total_budget left no time to wait for a ticker flush still in flight, and the final flush was skipped for a flush that was about to finish. main's supervisor budget grows by the same window.
The UPSERT keeps alias_for, so registering an alias name produced a row that archived in-process but was filtered out of every restore query (alias_for IS NULL) and vanished at the next restart. The HTTP handler already refuses this; the registry now does too, for every caller.
A re-archive that found another native type at the IOC flipped the registry type while the current partition kept its header type, so every append was refused until that partition rolled. Only changeTypeForPV (import_pv, after converting the stored partitions) may change a PV's type.
The ETL skips paused PVs only when a run starts, so a move that began before the pause could delete the source partition after the conversion opened it; the converted rewrite then resurrected the partition next to the coarser tier's old-type copy. EtlExecutor now exposes its move gate and the handler takes every distinct gate in chain order for the duration of convert_pv_type.
…less of subscribe order main subscribed the HTTP graceful-shutdown trigger only just before serving, and watch::Sender::subscribe marks the current value as seen, so a write_loop that died during ETL/cleanup setup never stopped the server. wait_for(|v| *v) resolves on state, not on an edge.
A PV whose buffered bytes were lost at flush (flush_ingest_writes failed, or a dirty-writer eviction via take_loss_markers) had already counted those samples in events_stored, and the flush owner only logged the loss. PendingReports now carries the appending task's counters so remove_failed can attribute the loss; getPVStatus exposes it as flushLosses and archiver_storage_flush_losses_total counts the reports.
…cross write_shards With write_shards > 1 the dispatcher drains the 500k main channel instantly into 4096-slot shard channels and drops on Full, so enabling shards cut the burst the engine absorbs before losing samples from 500k to 4096 per shard. Unset now means auto_per_shard_buffer(shards); an explicit value still overrides. Worst-case memory stays at the main channel's order (the shard total equals its capacity).
…protocol The CA path registered a DBF_DOUBLE waveform as ScalarDouble because dbr_field_to_arch_type never saw the element count, while epics-ca-rs decodes every count > 1 event as an array, so the per-sample type gate (and before it, encode_sample) refused every CA array sample. One rule now lives in ArchDbType::with_element_count and the registry applies it to every row it writes; init_schema promotes legacy rows once, which needs no partition conversion because no scalar-header file was ever created for them. A scalar channel re-archived with element_count > 1 is now a refused type change (its samples would be waveforms).
…_to_archiver CA autosize subscriptions carry the record's current NORD, and a count of 1 decodes as the scalar variant, so a waveform PV holding one element produced scalar samples that the shard type gate dropped. The converter now takes the registered ArchDbType and wraps a scalar into its one-element vector (ArchiverValue::into_vector, the same pairing as ArchDbType::with_element_count).
The registry promotes any type imported with element_count > 1 to the waveform form, so converting an array PV's partitions to a scalar type left the files and the registry row disagreeing and every later append failing the partition header check. Apply the same ArchDbType::with_element_count rule before the conversion starts.
The decoder leaves string arrays untyped, and both pv_field_to_arch_db_type and pv_field_scalar_to_archiver inferred the element type from the first element, so an empty string array could neither register nor be stored. The channel FieldDesc names the element type; the first element is only the fallback when it does not.
Two producers of uncounted shutdown loss. Shards drained on the same global flag as the dispatcher, so a shard could empty its channel and exit while the dispatcher was still moving the main queue's tail into it; the dispatcher now owns a per-shard drain signal (ShardSlot) that it flips only after that move. And nothing stopped the PV tasks, which kept sending into the pool during the drain and lost those samples when the receiver went away; ChannelManager::shutdown cancels every tracked producer, waits for them, and refuses later starts, and the binary runs it before flipping the pool's shutdown flag.
A resumed or restarted task is handed the current value again (CA initial monitor event, PVA initial get) with the timestamp of the sample already on disk; the gate started at zero and only rejected older timestamps, so that value was stored twice. task_counters seeds it from the registry and the last stored event, and an equal timestamp is now dropped as the same event redelivered. The CA tests wait for the connect-time event before putting: the in-process server can report a put that races the subscribe as the new value under the previous timestamp.
Seeding the ordering gate from the store cost a partition tail read per PV and per tier on the restore path, and a full-cache flush per call. A PV's counters now outlive its task, so a resume continues the high-water exactly; the first task in a process is seeded from the registry's committed last_timestamp alone, which the flush owner advances only for bytes on disk. destroy_pv discards the entry.
pva_handle_event runs on the pvAccess reactor task and dropped every sample while the queue was full, so a PV that changed once during a write stall lost that change until its next one; the CA producer waits instead, and the CA client coalesces to the latest value. The slot keeps the newest refused sample per PV, drain_overflow_slot delivers it with backpressure, and only a replaced parked sample counts as an overflow drop.
A failed write leaves every unwritten byte in the BufWriter, yet the flush evicted the writer and counted up to 64 KiB of samples per PV as lost on a transient EIO or ENOSPC. The write step now reports the PV deferred, the same "still buffered, retry next cycle" the owner already handles for a busy slot; a failed fsync still evicts, since the kernel drops the pages it reported an error for.
pause_pv only cancelled the token, so renamePV, changeTypeForPV, reassignAppliance and deletePV acted while the PV's tail could still be queued or inside an append: the late sample was stranded under the old name, refused by the converted partition, or deleted unmigrated. PvSample now carries an InFlight guard on PvCounters::in_flight, so every exit path settles by ownership, and PvHandle tracks its tasks; pause/stop/destroy wait for both, bounded by QUIESCE_TIMEOUT, and leave the registry status untouched on timeout.
…d receivePVMigration import_pv_with_protocol used INSERT OR REPLACE, which re-inserts the row without last_timestamp, so putPVTypeInfo and changeTypeForPV on an archived PV dropped its high-water and the next restart stored the connect-time redelivery again. renamePV left the destination row at NULL although the moved partitions end at the source's, and the migration receiver appends past the write pool's flush owner, so its row stayed NULL too. The import is now an upsert over the columns it owns; the two handlers commit the tail through update_last_timestamp, the migration one only after its flush succeeded.
…t waits stop_tasks now waits for a PV's tasks, so a wait that is bounded but not raced against the cancel token turns into pause latency on a disconnected PV: 5 s for the scan connect wait, 10 s for an extra field's, 30 s for a CA get, 5 s for the PVA client. pvmonitor_handle stays unraced on purpose: dropping it mid-subscribe could leave a server-side monitor running with no handle to stop it.
Nothing in the manager reads it: every write goes through the sample channel to the write pool, which owns the StoragePlugin. Removing it takes the storage parameter off ChannelManager::new and new_with_drift and the #[allow(dead_code)] that hid the dead field.
Both read the active-channel map, so a pause removed the PV from the drop, lost-connection and event-rate reports and from pvStatusAction, which is where an operator looks for why it was paused. The counters map already lives for the process (task_counters) and is cleared by destroy_pv, which is the Java lifetime of a stopped channel's stats.
It was the one producer await left outside the cancel select on the belief that dropping it mid-subscribe could leak a server-side monitor. epics-pva-rs 0.28.0 shows otherwise: the call awaits only the channel lookup, and op_monitor_handle spawns the subscription task synchronously before returning the handle, so nothing outlives a drop.
stop_tasks removed the channel and cancelled the producers, then awaited the quiesce before the caller set the registry status, so a caller that hit QUIESCE_TIMEOUT or was dropped mid-wait (the HTTP client gone) left the registry Active for a PV that no longer archived, silently in the second case. stop_and_finalize spawns the wait and the status commit as a tracked task holding the owned lock guard; the caller only waits on that task, bounded, and the status lands when the PV's queue drains.
…_gate task_counters seeded a PV's first gate of the process from the registry's committed last_timestamp, which can sit on either side of the store: behind it after a crash between a flush and the owner's commit, so the connect-time redelivery was stored twice, and ahead of it after a power loss with fsync_on_flush off, so that redelivery, the one sample still recoverable, was dropped. The owning shard now reads the store's tail once, at the PV's first sample, off the restore path. PlainPB's read-side flush is per PV for that, so a seed no longer flushes every other writer, the cost 1360bee moved away from.
The tail read ran inline on the shard's runtime worker with no bound, while the append is on the blocking pool under append_timeout so a stalled store parks a blocking thread. The read now takes the same route and bound; on timeout the registry seed stands and the PV counts as seeded, so the shard does not block on the store again per sample.
archive_pv, resume_pv and stop_and_finalize took the per-PV lock with no bound, so while a stop transition drained a stalled store the next operation on that PV queued for the whole stall, although the first caller had been given an error after 60 s. The same bound now applies to the lock wait, with an error naming the operation in progress.
Its CA server takes the record lock for subscription snapshots, so an initial event no longer reports a racing put's value under the pre-put timestamp. Stack::archive keeps waiting for the connect-time event: the live tests expect the seed value stored first.
Every caller discarded the refused sample, so the 152-byte PvSample in the Err variant only tripped clippy::result_large_err on rustc 1.98.
…seed TieredStorage::get_last_known_event honors SKIP_<TIER>_FOR_RETRIEVAL, so a gate seeded while a tier was routed around read an older tier's tail and stored the connect-time redelivery a second time. The stored view walks the tiers without the flags; seed_ordering_gate reads that.
…lize Under tokio's paused clock the 60 s elapse as soon as the runtime is idle, so the bounds are checked at their real value; without either timeout the matching test hangs.
Bump the workspace, the four internal-dep pins and the decoupled binary package to 0.4.2 (aligned again: epics-rs 0.20.4 → 0.28.1 touches every crate) and record the release in CHANGELOG.md.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fifty-four commits from the data-saving review rounds, one per finding, plus the epics-rs 0.28.1 bump whose CA server no longer emits a torn initial event. The one open item is the fsync_on_flush default, which stays off.