Skip to main content

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}