bacnet_server/server/local_writes.rs
1use super::cov_notify_context::CovNotifyContext;
2use super::*;
3use crate::command_lists::TakenRuns;
4use crate::handlers::{WriteCommitObserver, WriteTarget};
5use bacnet_objects::staging::StagingWritePlan;
6use bacnet_types::constructed::BACnetRecipient;
7
8#[path = "local_write_finish.rs"]
9mod finish;
10
11#[path = "local_access_inputs.rs"]
12mod access_inputs;
13pub(super) use finish::Committed;
14pub(super) use finish::local_runtime;
15
16#[cfg(test)]
17#[path = "input_present_value_tests.rs"]
18mod input_present_value_tests;
19
20#[cfg(test)]
21#[path = "staging_local_writes_tests.rs"]
22mod staging_local_writes_tests;
23
24#[cfg(test)]
25#[path = "local_array_index_tests.rs"]
26mod local_array_index_tests;
27
28#[cfg(test)]
29#[path = "local_index_staged_release_tests.rs"]
30mod local_index_staged_release_tests;
31
32/// What a local mutation is, and on whose behalf.
33///
34/// Inputs and noncommandable Values distinguish application updates from
35/// network-equivalent writes, including their Out_Of_Service ownership checks.
36#[derive(Debug, Clone, Copy, PartialEq)]
37pub(super) enum LocalWrite<'s> {
38 /// A trusted local program performing a network-equivalent property write.
39 Property {
40 property: PropertyIdentifier,
41 array_index: Option<u32>,
42 priority: Option<u8>,
43 },
44 /// The application supplying a supported object's logical `Present_Value`.
45 ApplicationPresentValue,
46 /// The application supplying a Loop's measured `Controlled_Variable_Value`.
47 ApplicationControlledVariableValue,
48 /// The application supplying one sample to an Averaging object.
49 ApplicationAveragingSample,
50 /// The application reporting an Averaging sample attempt that produced no
51 /// value.
52 ApplicationAveragingMiss,
53 /// The application supplying a Life Safety object's `Tracking_Value`.
54 ApplicationTrackingValue,
55 /// The application reporting an access-control object's input: an
56 /// access event, a credential read or a door's hardware state (#1132).
57 ApplicationAccessInput(&'s bacnet_objects::access_control::AccessControlInput),
58 /// One value of an inbound WriteGroup for a Channel's `Present_Value`, at
59 /// the priority the request gave it (Clause 15.11). Under the write guard
60 /// the Channel takes it only while its Control_Groups still holds `group`
61 /// and its Channel_Number is still `number`; otherwise the write is
62 /// skipped and queues nothing. With `inhibit_delay`, the run it queues
63 /// loses its delays when Allow_Group_Delay_Inhibit is TRUE at that moment.
64 /// It comes from the network: `requester`, the device that sent the
65 /// WriteGroup, is the source its Audit record names, with no invoke ID
66 /// (#1318).
67 WriteGroup {
68 group: u32,
69 number: u16,
70 priority: u8,
71 inhibit_delay: bool,
72 requester: &'s BACnetRecipient,
73 },
74}
75
76impl<T: TransportPort + 'static> BACnetServer<T> {
77 /// Arm or rearm a Life Safety object from trusted local application logic.
78 ///
79 /// This uses the object-internal state channel under the database write
80 /// lock. Network WriteProperty and WritePropertyMultiple remain unable to
81 /// forge `Operation_Expected`. After the lock is released, an actual
82 /// `Operation_Expected` readback change uses the exact Life Safety COV path;
83 /// rearming to the current value emits no notification.
84 ///
85 /// Like [`write_local`](Self::write_local), it must be awaited inside a
86 /// Tokio runtime, failing before anything changes outside one. Once the
87 /// object holds the operation, the timestamped capture and the COV
88 /// fanout the change owes run as a task of their own in the server's
89 /// request task set, which this future only waits for, so a caller
90 /// dropped then skips none of it (#1520). `stop()` aborts that task with
91 /// the other request tasks.
92 pub async fn set_life_safety_operation_expected_local(
93 &self,
94 oid: &ObjectIdentifier,
95 operation: LifeSafetyOperation,
96 ) -> Result<(), Error> {
97 let runtime = local_runtime()?;
98 self.active_network()?;
99 // Owned, so the fanout can take it to a task of its own with no wait
100 // between the change and the hand-off.
101 let mut db = Arc::clone(&self.db).write_owned().await;
102 let mut commit = crate::committed_cov::BackgroundCommit::new();
103 commit.before_change(&db, *oid);
104 let object = db.get_mut(oid).ok_or(Error::protocol(
105 ErrorClass::OBJECT,
106 ErrorCode::UNKNOWN_OBJECT,
107 ))?;
108 object.set_life_safety_operation_expected_internal(operation)?;
109 // Only a Life Safety Point or Zone reports the change. Another object
110 // whose setter takes the operation owes no COV, as before (#1520).
111 if crate::life_safety_cov::is_life_safety_object(*oid) {
112 commit.changed(*oid);
113 }
114 let fanout =
115 super::cov_fanout::CovFanout::new(&self.local_cov_context(), &self.event_suppressions);
116 self.finish_in_task(&runtime, async move {
117 // Under the change's guard: the database, then the COV table.
118 let committed = commit.finish(&fanout.db, &mut db, &fanout.cov_table).await;
119 drop(db);
120 fanout.fire(&committed).await;
121 })
122 .await
123 }
124
125 /// Write a property on a local object and fire the same post-write COV
126 /// and event notifications that a network [`WriteProperty`] does. When target
127 /// Audit reporting is configured, the same observer records eligible local
128 /// writes with local Device provenance and no invoke ID; Device recipient
129 /// changes retain their sole old/new pair owner. Eligible actual AV/BV Audit
130 /// policy changes reserve immediate delivery before assignment; unavailable
131 /// resources return SERVICES/SERVICE_REQUEST_DENIED without a policy change.
132 ///
133 /// This is the server-owned local-mutation entry point: it performs the
134 /// write under the database lock — routing `OBJECT_NAME` through the name
135 /// uniqueness check and index refresh, exactly like the network handler —
136 /// then releases the lock and runs the COV/event trigger path so a
137 /// subscription observes a local mutation just as it would a network one.
138 /// `source` explicitly selects the Device or an existing initiating object.
139 /// Tracked commands require a concrete selected Device; the selected Device
140 /// retains correction ownership regardless of the local initiator. Unrelated
141 /// properties preserve their behavior without a usable command origin.
142 /// Low-level object setters deliberately bypass this notification owner.
143 ///
144 /// A NULL to a property that isn't commandable and has no NULL in its
145 /// datatype succeeds and changes nothing, as over the network (#1396):
146 /// Audit records it, and no COV, event or Schedule work follows it.
147 ///
148 /// An array index is checked first, as the WriteProperty handler checks
149 /// it (#1426): UNKNOWN_PROPERTY for a property the object doesn't hold,
150 /// PROPERTY_IS_NOT_AN_ARRAY for one that isn't an array. Neither the
151 /// object nor Audit sees such a write. The writes a Command or Channel
152 /// run makes here get the same check.
153 ///
154 /// # Cancellation
155 ///
156 /// Dropped before the write commits, the future makes no change. Once it
157 /// has committed, the work the write owes (the event pass, the COV
158 /// fanout, a re-evaluated Schedule's target fanout, a Staging plan, and
159 /// the Command or Channel runs it started) runs as a task of its own in
160 /// the server's request task set, which this future only waits for, so
161 /// dropping the future skips none of it (#1367): subscribers hear of the
162 /// change, and a run ends as it would have, reported to its property
163 /// subscribers. `stop()` aborts that task with the other request tasks,
164 /// as it does a request handler: what it hadn't sent by then is not
165 /// sent, and a run it hadn't started ends as if none of its writes were
166 /// made (#1324), In_Process FALSE with each command unsuccessful, or
167 /// Write_Status FAILED. The call still returns `Ok(())` then: the write
168 /// was made. A panic in that work is raised again in the caller, as it
169 /// would be were the work done in place.
170 ///
171 /// The future must be awaited inside a Tokio runtime, which the task is
172 /// spawned onto: outside one it fails with [`Error::Encoding`] before
173 /// anything is written.
174 ///
175 /// [`WriteProperty`]: bacnet_services::write_property::WritePropertyRequest
176 pub async fn write_local(
177 &self,
178 oid: &ObjectIdentifier,
179 property: PropertyIdentifier,
180 array_index: Option<u32>,
181 value: PropertyValue,
182 priority: Option<u8>,
183 source: crate::LocalCommandSource,
184 ) -> Result<(), Error> {
185 self.write_local_as(
186 oid,
187 LocalWrite::Property {
188 property,
189 array_index,
190 priority,
191 },
192 value,
193 Some(source),
194 )
195 .await
196 }
197
198 /// Supply a supported object's logical `Present_Value` from Rust application code.
199 ///
200 /// Analog, Binary and Multi-state Inputs, noncommandable Values, Loop
201 /// (the control algorithm's output) and Life Safety Point and Zone opt in.
202 /// Binary Input values are logical INACTIVE/ACTIVE states after Polarity,
203 /// not raw physical states. The update runs the existing post-write
204 /// intrinsic-event and COV processing after the database lock is released;
205 /// configured delays and distribution policy still control delivery, and a
206 /// COV_Increment still filters Present_Value reports.
207 ///
208 /// Applications are denied while `Out_Of_Service` is TRUE to protect a
209 /// client's simulation value. This is local policy for Inputs and required
210 /// by the object clauses for the supported Values and Loop. NULL is an
211 /// invalid application value. Other object families fail closed; use
212 /// [`BACnetServer::write_local`] for network-equivalent writes and sourced
213 /// commands on commandable objects. The Python binding exposes this as
214 /// `BACnetServer.set_present_value_local`. A noncommandable Value that
215 /// tracks its source (`set_value_source_tracking`, #1552) then publishes
216 /// this server's Device as Value_Source, as Clause 19.5 asks of a write
217 /// the local device makes.
218 ///
219 /// A Life Safety Point or Zone takes an Enumerated BACnetLifeSafetyState,
220 /// a standard value or one from 256 to 65535, in or out of service, since
221 /// clients never write its Present_Value. The update leaves
222 /// Tracking_Value, Silenced and Operation_Expected alone, so any latching
223 /// until reset is the application's (see
224 /// [`BACnetServer::set_tracking_value_local`]). A change notifies
225 /// SubscribeCOV and Present_Value property subscribers through the Life
226 /// Safety COV path. The built-in objects run no intrinsic reporting, so
227 /// the post-write event pass raises nothing for them.
228 ///
229 /// Like [`write_local`](Self::write_local), it must be awaited inside a
230 /// Tokio runtime, failing before anything is written outside one, and a
231 /// caller dropped once the write has committed skips none of the work
232 /// the write owes (#1367).
233 ///
234 /// [`BACnetObject::set_present_value_internal`]: bacnet_objects::traits::BACnetObject::set_present_value_internal
235 pub async fn set_present_value_local(
236 &self,
237 oid: &ObjectIdentifier,
238 value: PropertyValue,
239 ) -> Result<(), Error> {
240 self.write_local_as(oid, LocalWrite::ApplicationPresentValue, value, None)
241 .await
242 }
243
244 /// Supply a Loop's measured `Controlled_Variable_Value` from the
245 /// application that runs its control algorithm.
246 ///
247 /// The value is a finite REAL: another datatype fails with PROPERTY /
248 /// INVALID_DATA_TYPE and NaN or an infinity with PROPERTY /
249 /// VALUE_OUT_OF_RANGE. An unknown object fails with OBJECT /
250 /// UNKNOWN_OBJECT and any object other than a Loop with OBJECT /
251 /// OPTIONAL_FUNCTIONALITY_NOT_SUPPORTED, the same errors as
252 /// [`BACnetServer::set_present_value_local`]. Unlike Present_Value it is
253 /// accepted while `Out_Of_Service` is TRUE, because Out_Of_Service
254 /// decouples only the output and Reliability from the algorithm.
255 ///
256 /// The update runs the same post-write COV processing as
257 /// `set_present_value_local` once the database lock is released. A
258 /// SubscribeCOVProperty on `Controlled_Variable_Value` is notified of the
259 /// change; a SubscribeCOV on the Loop is not, since the Loop's COV report
260 /// carries this value without being triggered by it, so its next report
261 /// carries the new value. The property stays read-only over the network.
262 ///
263 /// Like [`write_local`](Self::write_local), it must be awaited inside a
264 /// Tokio runtime, failing before anything is written outside one, and a
265 /// caller dropped once the write has committed skips none of the work
266 /// the write owes (#1367).
267 pub async fn set_controlled_variable_value_local(
268 &self,
269 oid: &ObjectIdentifier,
270 value: PropertyValue,
271 ) -> Result<(), Error> {
272 self.write_local_as(
273 oid,
274 LocalWrite::ApplicationControlledVariableValue,
275 value,
276 None,
277 )
278 .await
279 }
280
281 /// Record one sample for an Averaging object, taken by the application.
282 ///
283 /// The server samples an object that holds an `Object_Property_Reference`
284 /// itself, every Window_Interval / Window_Samples seconds (#1144). For an
285 /// object without one, the application samples whatever it averages,
286 /// spacing its readings that far apart, and passes each result here. On an
287 /// object the server samples, a call is one more attempt in the window
288 /// and doesn't move the server's schedule. A value may be
289 /// a BOOLEAN (FALSE and TRUE count as 0 and 1), Signed, Unsigned,
290 /// Enumerated or finite REAL; the object keeps its statistics in REAL
291 /// (Clause 12.5). `None` records an attempt that produced no value, such
292 /// as a failed read: it counts toward Attempted_Samples but not
293 /// Valid_Samples. Another datatype, Double included, fails with PROPERTY /
294 /// INVALID_DATA_TYPE and NaN or an infinity with PROPERTY /
295 /// VALUE_OUT_OF_RANGE, and a refused sample counts as neither attempted
296 /// nor valid. An unknown object fails with OBJECT / UNKNOWN_OBJECT and any
297 /// object other than an Averaging object with OBJECT /
298 /// OPTIONAL_FUNCTIONALITY_NOT_SUPPORTED, as with
299 /// [`BACnetServer::set_present_value_local`].
300 ///
301 /// Each call fills the next slot of the object's window, dropping the
302 /// oldest once Window_Samples are held, and Minimum_Value, Maximum_Value,
303 /// Average_Value, Attempted_Samples and Valid_Samples change together
304 /// under the database lock; the server's COV processing runs once the
305 /// lock is released. Averaging has no Table 13-1 row, so SubscribeCOV on
306 /// it is refused, but SubscribeCOVProperty and
307 /// SubscribeCOVPropertyMultiple are admitted: a numeric property is
308 /// reported when it moves by the subscription's COV increment, or on any
309 /// change when the subscription gives none (Table 13-1a). A move to or
310 /// from the NaN or infinite value of an empty window is always reported,
311 /// and staying at it never is. The report has no Status_Flags because the
312 /// object has none. The Python binding exposes this as
313 /// `BACnetServer.add_averaging_sample_local`.
314 ///
315 /// Like [`write_local`](Self::write_local), it must be awaited inside a
316 /// Tokio runtime, failing before anything is written outside one, and a
317 /// caller dropped once the write has committed skips none of the work
318 /// the write owes (#1367).
319 pub async fn add_averaging_sample_local(
320 &self,
321 oid: &ObjectIdentifier,
322 sample: Option<PropertyValue>,
323 ) -> Result<(), Error> {
324 match sample {
325 Some(value) => {
326 self.write_local_as(oid, LocalWrite::ApplicationAveragingSample, value, None)
327 .await
328 }
329 None => {
330 self.write_local_as(
331 oid,
332 LocalWrite::ApplicationAveragingMiss,
333 PropertyValue::Null,
334 None,
335 )
336 .await
337 }
338 }
339 }
340
341 /// Supply a Life Safety Point's or Zone's `Tracking_Value`, the live state
342 /// the application derived, while the server holds the object.
343 ///
344 /// The value is an Enumerated naming a standard BACnetLifeSafetyState or
345 /// one from the proprietary range 256..=65535. A reserved or larger number
346 /// fails with PROPERTY / VALUE_OUT_OF_RANGE and another datatype with
347 /// PROPERTY / INVALID_DATA_TYPE. An unknown object fails with OBJECT /
348 /// UNKNOWN_OBJECT and any object other than a Life Safety Point or Zone
349 /// with OBJECT / OPTIONAL_FUNCTIONALITY_NOT_SUPPORTED, as with
350 /// [`BACnetServer::set_present_value_local`]. Present_Value, Silenced and
351 /// Operation_Expected are left as they are: the object derives nothing
352 /// from Tracking_Value.
353 ///
354 /// While `Out_Of_Service` is TRUE the value replaces the one set aside for
355 /// the return to service, and a client's simulated Tracking_Value keeps
356 /// being served (#1108); the return to service then serves the latest
357 /// application value and notifies its subscribers. In service, a change
358 /// reaches SubscribeCOVProperty and SubscribeCOVPropertyMultiple
359 /// subscribers of Tracking_Value through the Life Safety COV path once the
360 /// database lock is released; whole-object SubscribeCOV reports don't
361 /// carry it. The property stays read-only over the network in service.
362 /// The Python binding exposes this as `BACnetServer.set_tracking_value_local`.
363 ///
364 /// Like [`write_local`](Self::write_local), it must be awaited inside a
365 /// Tokio runtime, failing before anything is written outside one, and a
366 /// caller dropped once the write has committed skips none of the work
367 /// the write owes (#1367).
368 pub async fn set_tracking_value_local(
369 &self,
370 oid: &ObjectIdentifier,
371 value: PropertyValue,
372 ) -> Result<(), Error> {
373 self.write_local_as(oid, LocalWrite::ApplicationTrackingValue, value, None)
374 .await
375 }
376
377 /// Commit the write, then hand what it owes to a task of its own in the
378 /// request task set and wait for that task (#1367). The commit holds the
379 /// database guard to the hand-off with no wait in between, so a caller
380 /// dropped once the write has committed leaves the task running: its
381 /// event pass, COV fanout, Schedule fanout, Staging plan and runs all go
382 /// ahead. `stop()` aborts the task with the other request tasks. With
383 /// no Tokio runtime to run that task on, nothing is written.
384 async fn write_local_as(
385 &self,
386 oid: &ObjectIdentifier,
387 write: LocalWrite<'_>,
388 value: PropertyValue,
389 source: Option<crate::LocalCommandSource>,
390 ) -> Result<(), Error> {
391 let runtime = local_runtime()?;
392 self.active_network()?;
393 let Some(committed) = self
394 .local_writer()
395 .commit(oid, write, value, source)
396 .await?
397 else {
398 return Ok(());
399 };
400 let runner = super::command_runs::CommandRunner::for_server(self);
401 // A closed set (the server is stopping) drops the task, and with it
402 // the guard and the runs, which then end unsuccessful (#1324).
403 self.finish_in_task(&runtime, async move {
404 let runs = runner.writer().finish(committed).await;
405 if !runs.is_empty() {
406 runner.start(runs);
407 }
408 })
409 .await
410 }
411
412 /// Borrow the handles a local write uses, once the caller has checked
413 /// that the network is still active.
414 pub(super) fn local_writer(&self) -> LocalWriter<'_, T> {
415 LocalWriter {
416 db: &self.db,
417 network: self
418 .network
419 .as_ref()
420 .expect("running local mutation owns network"),
421 cov_table: &self.cov_table,
422 cov_in_flight: &self.cov_in_flight,
423 notification_transactions: &self.notification_transactions,
424 comm_state: &self.comm_state,
425 learned_routers: &self.learned_routers,
426 device_bindings: &self.device_bindings,
427 event_suppressions: &self.event_suppressions,
428 config: &self.config,
429 }
430 }
431
432 /// Borrow the COV notification handles for a local mutation.
433 fn local_cov_context(&self) -> CovNotifyContext<'_, T> {
434 self.local_writer().cov_context()
435 }
436
437 /// Borrow the EventNotification handles for a local mutation.
438 fn local_event_delivery(&self) -> EventDelivery<'_, T> {
439 self.local_writer().event_delivery()
440 }
441
442 pub(super) async fn execute_initial_staging_plans(&self) {
443 let staging_oids = {
444 let database = self.db.read().await;
445 database.find_by_type(ObjectType::STAGING)
446 };
447 let staging_plans = {
448 let mut database = self.db.write().await;
449 Self::take_staging_plans(&mut database, &staging_oids)
450 };
451 Self::execute_staging_plans(
452 &self.local_event_delivery(),
453 &self.local_cov_context(),
454 staging_plans,
455 )
456 .await;
457 }
458
459 pub(super) fn take_staging_plans(
460 db: &mut ObjectDatabase,
461 oids: &[ObjectIdentifier],
462 ) -> Vec<StagingWritePlan> {
463 oids.iter()
464 .filter_map(|oid| {
465 db.get_mut(oid)
466 .and_then(|object| object.take_staging_write_plan_internal())
467 })
468 .collect()
469 }
470
471 /// Execute local Staging targets one mutation guard at a time.
472 ///
473 /// Every target mutation is generation-checked in the same database guard
474 /// that applies it. Event/COV work runs only after that guard is released.
475 pub(super) async fn execute_staging_plans(
476 delivery: &EventDelivery<'_, T>,
477 cov: &CovNotifyContext<'_, T>,
478 plans: Vec<StagingWritePlan>,
479 ) {
480 let db = delivery.db;
481 for plan in plans {
482 let mut all_succeeded = true;
483 for target in &plan.writes {
484 enum TargetResult {
485 Applied,
486 Failed,
487 Stale,
488 }
489
490 let result = {
491 let mut database = db.write().await;
492 let current = database
493 .get(&plan.source)
494 .and_then(|source| source.staging_generation_internal())
495 == Some(plan.generation);
496 if !current {
497 TargetResult::Stale
498 } else {
499 let origin = crate::command_source::resolve_local(
500 &database,
501 crate::LocalCommandSource::Object(plan.source),
502 )
503 .ok();
504 match database.get_mut(&target.object_identifier) {
505 Some(object) => match crate::command_source::write_target(
506 object,
507 PropertyIdentifier::PRESENT_VALUE,
508 None,
509 PropertyValue::Enumerated(u32::from(target.active)),
510 Some(plan.priority),
511 origin.as_ref(),
512 ) {
513 Ok(()) => {
514 // Timestamped references capture the
515 // target change under this guard (#856).
516 let capture = cov
517 .cov_table
518 .read()
519 .await
520 .timed_capture(target.object_identifier);
521 capture.run(&database);
522 TargetResult::Applied
523 }
524 Err(_) => TargetResult::Failed,
525 },
526 None => TargetResult::Failed,
527 }
528 }
529 };
530
531 match result {
532 TargetResult::Stale => break,
533 TargetResult::Failed => all_succeeded = false,
534 TargetResult::Applied => {
535 Self::fire_event_notifications_with_bindings(
536 delivery,
537 cov.cov_table,
538 &target.object_identifier,
539 )
540 .await;
541 Self::fire_cov_notifications(cov, &target.object_identifier).await;
542 }
543 }
544 }
545
546 let reliability_changed = {
547 let mut database = db.write().await;
548 let changed = database.get_mut(&plan.source).is_some_and(|source| {
549 source.complete_staging_write_plan_internal(plan.generation, all_succeeded)
550 });
551 if changed {
552 let capture = cov.cov_table.read().await.timed_capture(plan.source);
553 capture.run(&database);
554 }
555 changed
556 };
557 if reliability_changed {
558 Self::fire_event_notifications_with_bindings(delivery, cov.cov_table, &plan.source)
559 .await;
560 Self::fire_cov_notifications(cov, &plan.source).await;
561 }
562 }
563 }
564}
565
566/// The handles one local write borrows: the mutation under the database
567/// guard, then the event, COV and Staging work it owes once the guard is
568/// dropped. `write_local` writes through it, and so does each write of a
569/// Command object's run (Clause 12.10).
570pub(super) struct LocalWriter<'a, T: TransportPort + 'static> {
571 pub(super) db: &'a Arc<RwLock<ObjectDatabase>>,
572 pub(super) network: &'a Arc<NetworkLayer<T>>,
573 pub(super) cov_table: &'a Arc<RwLock<CovSubscriptionTable>>,
574 pub(super) cov_in_flight: &'a Arc<Semaphore>,
575 pub(super) notification_transactions: &'a Arc<NotificationTransactions>,
576 pub(super) comm_state: &'a Arc<CommState>,
577 pub(super) learned_routers: &'a Arc<Mutex<LearnedRouterCache>>,
578 pub(super) device_bindings: &'a Arc<RwLock<DeviceBindingTable>>,
579 pub(super) event_suppressions: &'a Arc<super::event_suppression::EventSuppressions>,
580 /// The server's config, which a run's task shares rather than copies
581 /// (#1521).
582 pub(super) config: &'a Arc<ServerConfig>,
583}
584
585impl<'a, T: TransportPort + 'static> LocalWriter<'a, T> {
586 /// The COV notification view of these handles.
587 pub(super) fn cov_context(&self) -> CovNotifyContext<'a, T> {
588 CovNotifyContext {
589 db: self.db,
590 network: self.network,
591 cov_table: self.cov_table,
592 cov_in_flight: self.cov_in_flight,
593 notification_transactions: self.notification_transactions,
594 comm_state: self.comm_state,
595 config: self.config,
596 }
597 }
598
599 /// The EventNotification delivery view of these handles.
600 pub(super) fn event_delivery(&self) -> EventDelivery<'a, T> {
601 EventDelivery {
602 db: self.db,
603 network: self.network,
604 comm_state: self.comm_state,
605 learned_routers: self.learned_routers,
606 notification_transactions: self.notification_transactions,
607 device_bindings: self.device_bindings,
608 suppressions: self.event_suppressions,
609 retry_timeout_ms: self.config.cov_retry_timeout_ms,
610 local_apdu_capacity: u32::from(self.config.device.max_apdu_length()),
611 }
612 }
613
614 /// Make one local write and the post-write work it owes. Returns the
615 /// Command and Channel runs it queued, directly or through a Schedule it
616 /// changed, for the caller to start. They are taken under the write's
617 /// guard and held in a [`TakenRuns`] until then, so dropping this future
618 /// part way, after the commit, ends them rather than leaving their
619 /// objects busy (#1324). `write_local` runs the two halves itself, the
620 /// second as a task of its own (#1367).
621 pub(super) async fn write(
622 &self,
623 oid: &ObjectIdentifier,
624 write: LocalWrite<'_>,
625 value: PropertyValue,
626 source: Option<crate::LocalCommandSource>,
627 ) -> Result<TakenRuns, Error> {
628 match self.commit(oid, write, value, source).await? {
629 Some(committed) => Ok(self.finish(committed).await),
630 None => Ok(TakenRuns::default()),
631 }
632 }
633
634 /// The first half of [`Self::write`]: the checks and the commit, under a
635 /// database guard it keeps. `None` for a write that changed nothing and
636 /// owes nothing more; otherwise the guard and what the write owes, for
637 /// [`Self::finish`]. Nothing here waits once the object has the value,
638 /// so a caller dropped before this returns made no change.
639 pub(super) async fn commit(
640 &self,
641 oid: &ObjectIdentifier,
642 write: LocalWrite<'_>,
643 value: PropertyValue,
644 source: Option<crate::LocalCommandSource>,
645 ) -> Result<Option<Committed>, Error> {
646 // Only a property write can carry OBJECT_NAME, so only it needs the name
647 // index kept in step.
648 let renaming = matches!(
649 write,
650 LocalWrite::Property { property, .. } if property == PropertyIdentifier::OBJECT_NAME
651 );
652 // A Pulse Converter's Input_Reference is judged as it commits (#1341).
653 let input_reference = matches!(
654 write,
655 LocalWrite::Property { property, .. } if property == PropertyIdentifier::INPUT_REFERENCE
656 );
657 let life_safety = crate::life_safety_cov::is_life_safety_object(*oid);
658 // A write the object saves first saves here, without the guard.
659 let durable = match write {
660 LocalWrite::Property {
661 property,
662 array_index,
663 ..
664 } => super::durable_writes::DurableTarget::local(*oid, property, array_index, &value),
665 _ => Vec::new(),
666 };
667 let staged = super::durable_writes::stage(self.db, durable).await;
668 {
669 // Owned, so the rest of the write can take it to a task of its
670 // own with no wait between the commit and the hand-off (#1367).
671 let mut db = Arc::clone(self.db).write_owned().await;
672 let snapshots = crate::life_safety_cov::LifeSafetyCovSnapshots::capture_oid(&db, *oid);
673 if let Err(error) = precheck(&db, oid, write, &value) {
674 // Nothing reached the object: what was staged goes back.
675 staged.release(&mut db);
676 return Err(error);
677 }
678 // A WriteGroup was planned under an earlier guard; the Channel
679 // may have left the group or changed number since.
680 let mut group_skips_delays = false;
681 if let LocalWrite::WriteGroup {
682 group,
683 number,
684 inhibit_delay,
685 ..
686 } = write
687 {
688 let object = db.get(oid).expect("existence checked above");
689 if !super::write_group::qualifies(object, group, number) {
690 debug!(
691 channel = %oid,
692 group,
693 number,
694 "WriteGroup skips a Channel no longer in the group with that number"
695 );
696 return Ok(None);
697 }
698 group_skips_delays =
699 inhibit_delay && super::write_group::allows_delay_inhibit(object);
700 }
701 let mut audit = match write {
702 LocalWrite::Property {
703 property,
704 array_index,
705 priority,
706 } => {
707 // Device's recipient sink exclusively owns the old/new pair.
708 let recipient_sink = property
709 == PropertyIdentifier::AUDIT_NOTIFICATION_RECIPIENT
710 && db
711 .get_mut(oid)
712 .and_then(|object| object.device_authority_internal())
713 .is_some();
714 if recipient_sink {
715 None
716 } else {
717 let mut audit = audit_reporter::WriteAudit::local(
718 self.config,
719 self.network,
720 self.notification_transactions,
721 &db,
722 );
723 let encoded = audit_reporter::small_value(&value).unwrap_or_default();
724 if let Some(audit) = &mut audit {
725 audit.before(
726 &db,
727 WriteTarget {
728 oid: *oid,
729 property,
730 array_index,
731 priority,
732 value: &encoded,
733 },
734 );
735 }
736 audit
737 }
738 }
739 // Audited as the WriteProperty of the Channel's Present_Value
740 // it amounts to, from the device that sent the WriteGroup.
741 LocalWrite::WriteGroup {
742 priority,
743 requester,
744 ..
745 } => {
746 let mut audit = audit_reporter::WriteAudit::requested_by(
747 self.config,
748 self.network,
749 self.notification_transactions,
750 requester.clone(),
751 None,
752 );
753 let encoded = audit_reporter::small_value(&value).unwrap_or_default();
754 audit.before(
755 &db,
756 WriteTarget {
757 oid: *oid,
758 property: PropertyIdentifier::PRESENT_VALUE,
759 array_index: None,
760 priority: Some(priority),
761 value: &encoded,
762 },
763 );
764 Some(audit)
765 }
766 LocalWrite::ApplicationPresentValue
767 | LocalWrite::ApplicationControlledVariableValue
768 | LocalWrite::ApplicationAveragingSample
769 | LocalWrite::ApplicationAveragingMiss
770 | LocalWrite::ApplicationTrackingValue
771 | LocalWrite::ApplicationAccessInput(_) => None,
772 };
773 let value = match write {
774 LocalWrite::Property { property, .. } => {
775 crate::local_references::localize(&db, *oid, property, value)
776 }
777 _ => value,
778 };
779 let null = crate::handlers::relinquish::is_null_value(&value);
780 let prepared = match write {
781 LocalWrite::Property {
782 property,
783 array_index,
784 priority,
785 } => {
786 let encoded = audit_reporter::small_value(&value).unwrap_or_default();
787 audit.as_mut().and_then(|audit| {
788 audit.commit_policy(
789 &mut db,
790 WriteTarget {
791 oid: *oid,
792 property,
793 array_index,
794 priority,
795 value: &encoded,
796 },
797 &value,
798 )
799 })
800 }
801 LocalWrite::ApplicationPresentValue
802 | LocalWrite::ApplicationControlledVariableValue
803 | LocalWrite::ApplicationAveragingSample
804 | LocalWrite::ApplicationAveragingMiss
805 | LocalWrite::ApplicationTrackingValue
806 | LocalWrite::ApplicationAccessInput(_)
807 | LocalWrite::WriteGroup { .. } => None,
808 };
809 // The application's own Present_Value comes from this Device, which
810 // a tracked noncommandable Value publishes as its source (#1552).
811 let source = source.or(matches!(write, LocalWrite::ApplicationPresentValue)
812 .then_some(crate::LocalCommandSource::ServerDevice));
813 let command_origin =
814 source.and_then(|source| crate::command_source::resolve_local(&db, source).ok());
815 let result = prepared.unwrap_or_else(|| {
816 let object = db.get_mut(oid).expect("existence checked above");
817 match write {
818 LocalWrite::Property {
819 property,
820 array_index,
821 priority,
822 } => {
823 crate::device_view::check_executor_owned_write(*oid, property)?;
824 crate::command_source::write_target(
825 object,
826 property,
827 array_index,
828 value,
829 priority,
830 command_origin.as_ref(),
831 )
832 }
833 LocalWrite::ApplicationPresentValue => match &command_origin {
834 Some(origin) => object.set_present_value_from_internal(value, origin),
835 None => object.set_present_value_internal(value),
836 },
837 LocalWrite::ApplicationControlledVariableValue => {
838 object.set_controlled_variable_value_internal(value)
839 }
840 LocalWrite::ApplicationAveragingSample => {
841 object.add_averaging_sample_internal(Some(value))
842 }
843 LocalWrite::ApplicationAveragingMiss => {
844 object.add_averaging_sample_internal(None)
845 }
846 LocalWrite::ApplicationTrackingValue => {
847 object.set_tracking_value_internal(value)
848 }
849 LocalWrite::ApplicationAccessInput(input) => {
850 object.report_access_input_internal(input.clone())
851 }
852 LocalWrite::WriteGroup { priority, .. } => object.write_property(
853 PropertyIdentifier::PRESENT_VALUE,
854 None,
855 value,
856 Some(priority),
857 ),
858 }
859 });
860 staged.release(&mut db);
861 // A NULL the property left as it was succeeds unchanged
862 // (`handlers::relinquish`): audited as the write it is, with no
863 // post-write work, since nothing changed.
864 if let (
865 Err(error),
866 LocalWrite::Property {
867 property,
868 array_index,
869 ..
870 },
871 ) = (&result, write)
872 {
873 let object = db.get(oid).expect("existence checked above");
874 if null
875 && crate::handlers::relinquish::leaves_unchanged(
876 object,
877 property,
878 array_index,
879 error,
880 )
881 {
882 if let Some(audit) = &mut audit {
883 audit.committed(&mut db);
884 }
885 return Ok(None);
886 }
887 }
888 if let Err(error) = result {
889 if let Some(audit) = &mut audit {
890 audit.failed(&mut db, &error);
891 }
892 return Err(error);
893 }
894 if let Some(audit) = &mut audit {
895 audit.committed(&mut db);
896 }
897 if renaming {
898 db.update_name_index(oid);
899 }
900 if input_reference {
901 db.check_input_reference(oid);
902 }
903 let staging_plans =
904 BACnetServer::<T>::take_staging_plans(&mut db, std::slice::from_ref(oid));
905 let mut command_runs = TakenRuns::take(self.db, &mut db, std::slice::from_ref(oid));
906 if group_skips_delays {
907 super::write_group::skip_delays(&mut command_runs, *oid);
908 }
909 let changes = snapshots.changes(&db, std::slice::from_ref(oid));
910 Ok(Some(Committed {
911 db,
912 oid: *oid,
913 life_safety,
914 changes,
915 staging_plans,
916 command_runs,
917 }))
918 }
919 }
920}
921
922/// The checks made under the write's guard before the audit hooks or the
923/// object see the value, in WriteProperty's order: the object exists, and
924/// for a property write, the array index fits the property (WriteProperty's
925/// own [`check_write_array_index`](crate::handlers::check_write_array_index))
926/// and a new Object_Name is free. The name check calls
927/// [`ObjectDatabase::check_name_available`] directly, which is what
928/// WriteProperty's `check_and_prepare_name_write` runs for a CharacterString.
929/// A Command's or Channel's write takes this path too; a Schedule's target
930/// writes run the same index gate in `schedule::deliver`.
931fn precheck(
932 db: &ObjectDatabase,
933 oid: &ObjectIdentifier,
934 write: LocalWrite<'_>,
935 value: &PropertyValue,
936) -> Result<(), Error> {
937 let object = db.get(oid).ok_or(Error::protocol(
938 ErrorClass::OBJECT,
939 ErrorCode::UNKNOWN_OBJECT,
940 ))?;
941 let LocalWrite::Property {
942 property,
943 array_index,
944 ..
945 } = write
946 else {
947 return Ok(());
948 };
949 crate::handlers::check_write_array_index(object, property, array_index)?;
950 match value {
951 PropertyValue::CharacterString(new_name) if property == PropertyIdentifier::OBJECT_NAME => {
952 db.check_name_available(oid, new_name)
953 }
954 _ => Ok(()),
955 }
956}