bacnet_server/server/lifecycle.rs
1use super::event_notifications::ResolvedIntrinsicTransition;
2use super::*;
3
4/// Resolve the configured Event Enrollment interval into a tick period.
5///
6/// `tokio::time::interval` panics on a zero period, and that panic would land
7/// inside a spawned task — `start` would still return `Ok` while enrollment
8/// evaluation was silently dead. A configured `0` is clamped to one second
9/// instead, matching how an invalid `vendor_id` is handled: warn loudly and
10/// keep the device running. Use `enable_event_enrollment(false)` to actually
11/// disable evaluation.
12pub(super) fn event_enrollment_period(secs: u64) -> Duration {
13 if secs == 0 {
14 warn!(
15 "event_enrollment_interval_secs is 0; clamping to 1s. \
16 Use enable_event_enrollment(false) to disable Event Enrollment evaluation"
17 );
18 return Duration::from_secs(1);
19 }
20 Duration::from_secs(secs)
21}
22
23impl<T: TransportPort + 'static> BACnetServer<T> {
24 pub(super) async fn start_with_clock_mode_and_bindings(
25 mut config: ServerConfig,
26 mut db: ObjectDatabase,
27 transport: T,
28 clock_config: Option<ClockConfig>,
29 configured_device_bindings: Vec<DeviceBinding>,
30 ) -> Result<Self, Error> {
31 // Validate every configured route against the concrete transport before
32 // mutating the database or starting network work.
33 let device_bindings =
34 DeviceBindingTable::from_configured(configured_device_bindings, |mac| {
35 transport.is_broadcast_mac(mac)
36 })?;
37 let transport_max = transport.max_apdu_length() as u32;
38 config.max_apdu_length = config.max_apdu_length.min(transport_max);
39 let max_apdu = u16::try_from(config.max_apdu_length).map_err(|_| {
40 Error::Encoding(format!(
41 "invalid max_apdu_length {}; expected one of 50, 128, 206, 480, 1024, 1476",
42 config.max_apdu_length
43 ))
44 })?;
45 validate_max_apdu_length(max_apdu)?;
46
47 if config.vendor_id == 0 {
48 warn!("vendor_id is 0 (ASHRAE reserved); set a valid vendor ID for production use");
49 }
50
51 let clock = clock_config.map(|config| Arc::new(ServerClock::new(config)));
52 let reader = clock
53 .as_ref()
54 .map(|clock| Arc::clone(clock) as Arc<dyn bacnet_objects::clock::ClockReader>);
55 db.set_clock_reader(reader);
56 let monotonic_origin = tokio::time::Instant::now();
57 let monotonic_clock: Arc<bacnet_objects::traits::MonotonicClock> = Arc::new(move || {
58 tokio::time::Instant::now().saturating_duration_since(monotonic_origin)
59 });
60 db.set_monotonic_clock_internal(Some(monotonic_clock));
61
62 let mut network = NetworkLayer::new(transport);
63 let apdu_rx = network.start().await?;
64 let local_mac = MacAddr::from_slice(network.local_mac());
65
66 let network = Arc::new(network);
67 let db = Arc::new(RwLock::new(db));
68 let cov_table = Arc::new(RwLock::new(CovSubscriptionTable::new()));
69 let seg_ack_senders: Arc<Mutex<HashMap<SegKey, Arc<SegmentedSendHandle>>>> =
70 Arc::new(Mutex::new(HashMap::new()));
71 let seg_send_permits = Arc::new(Semaphore::new(MAX_SEG_SENDERS));
72
73 let cov_in_flight = Arc::new(Semaphore::new(255));
74 let server_tsm = Arc::new(Mutex::new(ServerTsm::new()));
75 let notification_transactions = NotificationTransactions::new();
76 let confirmed_request_tracker = Arc::new(ConfirmedRequestTracker::default());
77 let device_bindings = Arc::new(RwLock::new(device_bindings));
78 let comm_state = Arc::new(AtomicU8::new(0)); // 0 = Enable (default)
79 let dcc_timer: Arc<Mutex<Option<JoinHandle<()>>>> = Arc::new(Mutex::new(None));
80
81 let network_dispatch = Arc::clone(&network);
82 let db_dispatch = Arc::clone(&db);
83 let cov_dispatch = Arc::clone(&cov_table);
84 let seg_ack_dispatch = Arc::clone(&seg_ack_senders);
85 let seg_send_permits_dispatch = Arc::clone(&seg_send_permits);
86 let cov_in_flight_dispatch = Arc::clone(&cov_in_flight);
87 let server_tsm_dispatch = Arc::clone(&server_tsm);
88 let notification_transactions_dispatch = Arc::clone(¬ification_transactions);
89 let confirmed_request_tracker_dispatch = Arc::clone(&confirmed_request_tracker);
90 let device_bindings_dispatch = Arc::clone(&device_bindings);
91 let comm_state_dispatch = Arc::clone(&comm_state);
92 let dcc_timer_dispatch = Arc::clone(&dcc_timer);
93 let config_dispatch = Arc::new(config.clone());
94 let clock_dispatch = clock.clone();
95
96 let dispatch_task = tokio::spawn(async move {
97 let mut apdu_rx = apdu_rx;
98 let mut seg_receivers: HashMap<SegKey, SegmentedRequestState> = HashMap::new();
99
100 while let Some(received) = apdu_rx.recv().await {
101 let now = Instant::now();
102 super::segmented_receive::expire_segmented_requests(&mut seg_receivers, now);
103
104 match apdu::decode_apdu(received.apdu.clone()) {
105 Ok(decoded) => {
106 let source_mac = received.source_mac.clone();
107 let source_network = received.source_network.clone();
108
109 // Clause 5.4.5.2 AbortPDU_Received: a peer's Abort
110 // ('server' = FALSE) ends any reassembly session for
111 // its transaction. A side effect, not a short
112 // circuit — the PDU still reaches `dispatch`, whose
113 // Abort arm cancels in-flight segmented response
114 // senders and records server-TSM results (#377).
115 if let Apdu::Abort(ref abt) = decoded {
116 if !abt.sent_by_server {
117 let key = segmented_transaction_key(
118 source_mac.as_slice(),
119 source_network.as_ref(),
120 abt.invoke_id,
121 );
122 seg_receivers.remove(&key);
123 }
124 }
125
126 let mut received = Some(received);
127 let handled = if let Apdu::ConfirmedRequest(ref req) = decoded {
128 if req.segmented {
129 let seq = req.sequence_number.unwrap_or(0);
130 let key = segmented_transaction_key(
131 source_mac.as_slice(),
132 source_network.as_ref(),
133 req.invoke_id,
134 );
135
136 // Clause 5.4.5.1
137 // ConfirmedSegmentedReceivedNotSupported: a
138 // device that does not support segmented
139 // reception answers segment traffic with this
140 // Abort instead of reassembling — the
141 // configured Segmentation value is the
142 // advertisement peers plan transfers around
143 // (#381).
144 let receives_segments = config_dispatch.segmentation_supported
145 == Segmentation::BOTH
146 || config_dispatch.segmentation_supported
147 == Segmentation::RECEIVE;
148 if !receives_segments {
149 Self::send_server_abort(
150 &network_dispatch,
151 &source_mac,
152 source_network.as_ref(),
153 req.invoke_id,
154 AbortReason::SEGMENTATION_NOT_SUPPORTED,
155 )
156 .await;
157 continue;
158 }
159
160 let mut ack_to_send: Option<SegmentAckPdu> = None;
161 let mut final_total: Option<usize> = None;
162
163 // The live session is consulted before the
164 // `seq == 0` open path: Clause 20.1.2.7 wraps
165 // the sequence number modulo 256, so segment
166 // 256 of a long request arrives as another
167 // `seq == 0` — treating it as a fresh initial
168 // segment would silently replace the session
169 // and reassemble only the tail (#364).
170 let saved_bytes =
171 super::segmented_receive::saved_request_payload_bytes(
172 &seg_receivers,
173 );
174 if let Some(state) = seg_receivers.get_mut(&key) {
175 // Clause 5.4.5.2 restarts SegmentTimer
176 // for accepted, duplicate and
177 // out-of-order segments alike, so the
178 // refresh precedes the ordering checks.
179 state.last_activity = Instant::now();
180 if seq != state.expected_seq {
181 ack_to_send =
182 super::segmented_receive::classify_non_next_segment(
183 state,
184 req.invoke_id,
185 seq,
186 );
187 } else {
188 // In-order NEW segment: duplicates
189 // and gaps returned above, so a
190 // retransmission can never trip the
191 // cap (Clause 5.4.5.2
192 // DuplicateSegmentReceived requires
193 // duplicates be discarded, not
194 // punished).
195 if state.accepted_segments >= MAX_REQUEST_SEGMENTS {
196 // Clause 5.4.5.2 has no overflow
197 // transition; SendAbort ('server'
198 // = TRUE, reason a local matter)
199 // is its one generic escape, and
200 // Clause 18.10's BUFFER_OVERFLOW
201 // — "a buffer capacity has been
202 // exceeded" — is the fit (#364).
203 warn!(
204 invoke_id = req.invoke_id,
205 accepted = state.accepted_segments,
206 "Segmented request exceeds reassembly capacity, aborting"
207 );
208 seg_receivers.remove(&key);
209 Self::send_server_abort(
210 &network_dispatch,
211 &source_mac,
212 source_network.as_ref(),
213 req.invoke_id,
214 AbortReason::BUFFER_OVERFLOW,
215 )
216 .await;
217 continue;
218 }
219 if let Err(e) = state.payload.save_new(
220 seq,
221 req.service_request.clone(),
222 saved_bytes,
223 ) {
224 // An unsaveable segment ends the
225 // session the same way — leaving
226 // it dangling told the peer
227 // nothing while this side could
228 // never complete (#364).
229 warn!(error = %e, "Rejecting unsaveable segment");
230 seg_receivers.remove(&key);
231 Self::send_server_abort(
232 &network_dispatch,
233 &source_mac,
234 source_network.as_ref(),
235 req.invoke_id,
236 AbortReason::BUFFER_OVERFLOW,
237 )
238 .await;
239 continue;
240 }
241 state.last_progress = Instant::now();
242 state.accepted_segments += 1;
243 state.expected_seq = seq.wrapping_add(1);
244 state.last_acked_seq = seq;
245 state.window_pos += 1;
246 let should_ack = !req.more_follows
247 || state.window_pos >= state.actual_window_size;
248 if should_ack {
249 state.window_pos = 0;
250 state.initial_sequence_number = state.last_acked_seq;
251 state.duplicate_count = 0;
252 ack_to_send = Some(SegmentAckPdu {
253 negative_ack: false,
254 sent_by_server: true,
255 invoke_id: req.invoke_id,
256 sequence_number: seq,
257 actual_window_size: state.actual_window_size,
258 });
259 }
260 if !req.more_follows {
261 // The count, not `seq + 1`: the
262 // wire sequence number is modulo
263 // 256 (Clause 20.1.2.7) and says
264 // nothing about how many segments
265 // were accepted (#364).
266 final_total = Some(state.accepted_segments);
267 }
268 }
269 } else if seq == 0 {
270 let proposed_window_size =
271 req.proposed_window_size.unwrap_or(0);
272 if !(1..=127).contains(&proposed_window_size) {
273 warn!(
274 invoke_id = req.invoke_id,
275 proposed_window_size,
276 "Rejecting segmented request with invalid proposed window size"
277 );
278 Self::send_server_abort(
279 &network_dispatch,
280 &source_mac,
281 source_network.as_ref(),
282 req.invoke_id,
283 AbortReason::WINDOW_SIZE_OUT_OF_RANGE,
284 )
285 .await;
286 continue;
287 }
288
289 // New sessions only; global capacity precedes peer quota.
290 if let Some(reason) =
291 super::segmented_receive::segmented_request_admission_error(
292 &seg_receivers,
293 &key,
294 )
295 {
296 Self::send_server_abort(
297 &network_dispatch,
298 &source_mac,
299 source_network.as_ref(),
300 req.invoke_id,
301 reason,
302 )
303 .await;
304 continue;
305 }
306
307 let mut payload =
308 super::segmented_receive::RequestPayload::new(req);
309 if let Err(e) = payload.save_new(
310 seq,
311 req.service_request.clone(),
312 saved_bytes,
313 ) {
314 // No session exists to drop on this
315 // path; the Abort is what tells the
316 // peer instead of leaving it to time
317 // out (#364).
318 warn!(error = %e, "Rejecting unsaveable segment");
319 Self::send_server_abort(
320 &network_dispatch,
321 &source_mac,
322 source_network.as_ref(),
323 req.invoke_id,
324 AbortReason::BUFFER_OVERFLOW,
325 )
326 .await;
327 continue;
328 }
329 let actual_window_size = proposed_window_size;
330 let mut state = SegmentedRequestState {
331 payload,
332 last_activity: Instant::now(),
333 last_progress: Instant::now(),
334 expected_seq: 1,
335 initial_sequence_number: 0,
336 duplicate_count: 0,
337 last_acked_seq: 0,
338 window_pos: 1,
339 actual_window_size,
340 accepted_segments: 1,
341 };
342 let should_ack =
343 !req.more_follows || state.window_pos >= actual_window_size;
344 if should_ack {
345 state.window_pos = 0;
346 state.initial_sequence_number = state.last_acked_seq;
347 state.duplicate_count = 0;
348 ack_to_send = Some(SegmentAckPdu {
349 negative_ack: false,
350 sent_by_server: true,
351 invoke_id: req.invoke_id,
352 sequence_number: seq,
353 actual_window_size,
354 });
355 }
356 if !req.more_follows {
357 final_total = Some(1);
358 }
359 seg_receivers.insert(key.clone(), state);
360 } else {
361 warn!(
362 invoke_id = req.invoke_id,
363 seq = seq,
364 "Received non-initial segment without prior segment 0, aborting"
365 );
366 Self::send_server_abort(
367 &network_dispatch,
368 &source_mac,
369 source_network.as_ref(),
370 req.invoke_id,
371 AbortReason::INVALID_APDU_IN_THIS_STATE,
372 )
373 .await;
374 continue;
375 }
376
377 if let Some(seg_ack) = ack_to_send {
378 let seg_ack = Apdu::SegmentAck(seg_ack);
379 let mut ack_buf = BytesMut::new();
380 encode_apdu(&mut ack_buf, &seg_ack)
381 .expect("valid APDU encoding");
382 if let Err(e) = Self::send_confirmed_response_apdu(
383 &network_dispatch,
384 &ack_buf,
385 &source_mac,
386 source_network.as_ref(),
387 )
388 .await
389 {
390 warn!(
391 error = %e,
392 "Failed to send SegmentAck for segmented request"
393 );
394 }
395 }
396
397 if let Some(total) = final_total {
398 if let Some(state) = seg_receivers.remove(&key) {
399 match state.payload.complete(total) {
400 Ok(reassembled) => {
401 debug!(
402 invoke_id = reassembled.invoke_id,
403 segments = total,
404 payload_len = reassembled.service_request.len(),
405 "Reassembled segmented ConfirmedRequest"
406 );
407 Self::dispatch(
408 &db_dispatch,
409 &network_dispatch,
410 &cov_dispatch,
411 &seg_ack_dispatch,
412 &seg_send_permits_dispatch,
413 &cov_in_flight_dispatch,
414 &server_tsm_dispatch,
415 ¬ification_transactions_dispatch,
416 &confirmed_request_tracker_dispatch,
417 &device_bindings_dispatch,
418 &comm_state_dispatch,
419 &dcc_timer_dispatch,
420 &config_dispatch,
421 &clock_dispatch,
422 &source_mac,
423 Apdu::ConfirmedRequest(reassembled),
424 received.take().unwrap_or_else(|| {
425 warn!("received consumed twice - using empty fallback");
426 bacnet_network::layer::ReceivedApdu {
427 apdu: bytes::Bytes::new(),
428 source_mac: bacnet_types::MacAddr::new(),
429 source_network: None,
430 link_layer_group: false,
431 is_group: false,
432 data_attributes: Vec::new(),
433 reply_tx: None,
434 }
435 }),
436 )
437 .await;
438 }
439 Err(e) => {
440 warn!(
441 error = %e,
442 "Failed to reassemble segmented request"
443 );
444 }
445 }
446 }
447 }
448
449 true
450 } else {
451 false
452 }
453 } else {
454 false
455 };
456
457 if !handled {
458 Self::dispatch(
459 &db_dispatch,
460 &network_dispatch,
461 &cov_dispatch,
462 &seg_ack_dispatch,
463 &seg_send_permits_dispatch,
464 &cov_in_flight_dispatch,
465 &server_tsm_dispatch,
466 ¬ification_transactions_dispatch,
467 &confirmed_request_tracker_dispatch,
468 &device_bindings_dispatch,
469 &comm_state_dispatch,
470 &dcc_timer_dispatch,
471 &config_dispatch,
472 &clock_dispatch,
473 &source_mac,
474 decoded,
475 received.take().unwrap_or_else(|| {
476 warn!("received consumed twice — using empty fallback");
477 bacnet_network::layer::ReceivedApdu {
478 apdu: bytes::Bytes::new(),
479 source_mac: bacnet_types::MacAddr::new(),
480 source_network: None,
481 link_layer_group: false,
482 is_group: false,
483 data_attributes: Vec::new(),
484 reply_tx: None,
485 }
486 }),
487 )
488 .await;
489 }
490 }
491 Err(e) => {
492 warn!(error = %e, "Server failed to decode received APDU");
493 }
494 }
495 }
496 });
497
498 let cov_table_for_purge = Arc::clone(&cov_table);
499 let cov_purge_task = tokio::spawn(async move {
500 let mut interval = tokio::time::interval(Duration::from_secs(30));
501 loop {
502 interval.tick().await;
503 let mut table = cov_table_for_purge.write().await;
504 let purged = table.purge_expired();
505 if purged > 0 {
506 debug!(purged, "Purged expired COV subscriptions");
507 }
508 }
509 });
510
511 let fault_detection_task = if config.enable_fault_detection {
512 let db_fault = Arc::clone(&db);
513 Some(tokio::spawn(async move {
514 let detector = crate::fault_detection::FaultDetector::default();
515 let mut interval = tokio::time::interval(Duration::from_secs(10));
516 loop {
517 interval.tick().await;
518 let mut db_guard = db_fault.write().await;
519 let changes = detector.evaluate(&mut db_guard);
520 for change in &changes {
521 debug!(
522 object = %change.object_id,
523 old = change.old_reliability,
524 new = change.new_reliability,
525 "Fault detection: reliability changed"
526 );
527 }
528 }
529 }))
530 } else {
531 None
532 };
533
534 let event_enrollment_task = if config.enable_event_enrollment {
535 let ee_period = event_enrollment_period(config.event_enrollment_interval_secs);
536 // The delay countdown converts seconds to passes with
537 // `ceil(delay / period)`, so the evaluator needs the actual,
538 // clamped interval — not the raw config value.
539 Some(
540 super::event_enrollment_lifecycle::spawn_event_enrollment_task(
541 Arc::clone(&db),
542 Arc::clone(&network),
543 Arc::clone(&comm_state),
544 Arc::clone(&server_tsm),
545 Arc::clone(¬ification_transactions),
546 Arc::clone(&device_bindings),
547 ee_period,
548 config.cov_retry_timeout_ms,
549 ),
550 )
551 } else {
552 None
553 };
554
555 let db_trend = Arc::clone(&db);
556 let trend_log_state: crate::trend_log::TrendLogState =
557 Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new()));
558 let trend_log_task = Some(tokio::spawn(async move {
559 let mut interval = tokio::time::interval(Duration::from_secs(1));
560 loop {
561 interval.tick().await;
562 crate::trend_log::poll_trend_logs(&db_trend, &trend_log_state).await;
563 }
564 }));
565
566 let db_schedule = Arc::clone(&db);
567 let network_schedule = Arc::clone(&network);
568 let cov_table_schedule = Arc::clone(&cov_table);
569 let cov_in_flight_schedule = Arc::clone(&cov_in_flight);
570 let notification_transactions_schedule = Arc::clone(¬ification_transactions);
571 let comm_state_schedule = Arc::clone(&comm_state);
572 let schedule_config = config.clone();
573 let schedule_tick_task = Some(tokio::spawn(async move {
574 let mut interval = tokio::time::interval(Duration::from_secs(60));
575 loop {
576 interval.tick().await;
577 let changes =
578 crate::schedule::tick_schedules_with_life_safety_cov(&db_schedule, 0).await;
579 for change in changes {
580 Self::fire_life_safety_cov_notifications(
581 &db_schedule,
582 &network_schedule,
583 &cov_table_schedule,
584 &cov_in_flight_schedule,
585 ¬ification_transactions_schedule,
586 &comm_state_schedule,
587 &schedule_config,
588 &change.object_identifier,
589 &change.changed_properties,
590 )
591 .await;
592 }
593 }
594 }));
595
596 // One-second intrinsic-reporting task: advances the `Time_Delay`
597 // countdown for any object with a pending delayed transition and sends
598 // the EventNotification when the delay elapses. The per-write path
599 // only *seeds* a pending transition (see `fire_event_notifications`);
600 // this task is the sole confirmer, so repeated writes cannot shorten
601 // the delay (ASHRAE 135-2020 §13.2.4). Runs unconditionally like the
602 // trend-log task — a no-pending tick is a cheap empty iteration.
603 //
604 // It is also what carries Reliability into event-state-detection. Per
605 // Clause 13.2.2 the FAULT determination is a standing condition, so each
606 // tick re-derives it from the object's current `Reliability` rather than
607 // reacting to a change event. That is why the fault detector above can
608 // keep merely *logging* its `ReliabilityChange` records: whoever writes
609 // Reliability — an object's opt-in evaluation hook, a local write, or a
610 // network write — reaches detection through this tick, and no route
611 // needs to notify anything. `enable_fault_detection` therefore governs
612 // only whether those object-owned hooks run every 10 seconds, never
613 // whether an existing Reliability is honored.
614 //
615 // Six of the nine wired object types have no route that can set
616 // Reliability, so the fault path is correct but inert on them (#218).
617 let db_intrinsic = Arc::clone(&db);
618 let network_intrinsic = Arc::clone(&network);
619 let comm_state_intrinsic = Arc::clone(&comm_state);
620 let server_tsm_intrinsic = Arc::clone(&server_tsm);
621 let notification_transactions_intrinsic = Arc::clone(¬ification_transactions);
622 let device_bindings_intrinsic = Arc::clone(&device_bindings);
623 let intrinsic_retry_ms = config.cov_retry_timeout_ms;
624 let intrinsic_reporting_task = Some(tokio::spawn(async move {
625 let mut interval = tokio::time::interval(Duration::from_secs(1));
626 // The countdown decrements exactly once per call, so a delayed wake
627 // must NOT burst-deliver missed ticks (each would decrement
628 // `remaining`, compressing the Time_Delay). `Delay` collapses a
629 // missed deadline into a single tick, preserving per-second
630 // granularity (ASHRAE 135-2020 §13.2.4).
631 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
632 loop {
633 interval.tick().await;
634 // DCC gates the outbound sender, not event-state detection or
635 // the local transition actions in Clause 13.2.2.1.4.
636 // Collect resolved transitions under a brief write lock, then
637 // drop it before sending (never hold the db lock across a
638 // network send — matches the per-write notification path).
639 //
640 // Event_Enable gates distribution only (Clause 12.12), so every
641 // built-in proposal is committed locally before a suppressed
642 // transition is omitted from the outbound work list. Legacy
643 // implementations already commit during their tick and bypass
644 // the atomic hook explicitly.
645 let fired = {
646 let mut db = db_intrinsic.write().await;
647 let mut out = Vec::new();
648 for oid in db.list_objects() {
649 let evaluated = db.get_mut(&oid).and_then(|object| {
650 let requires_atomic_commit =
651 object.intrinsic_reporting_requires_atomic_commit();
652 object
653 .tick_intrinsic_reporting()
654 .map(|outcome| (requires_atomic_commit, outcome))
655 });
656 let resolved = evaluated.and_then(|(requires_atomic_commit, outcome)| {
657 if requires_atomic_commit {
658 Self::commit_intrinsic_transition(&mut db, &oid, outcome)
659 .map(ResolvedIntrinsicTransition::Committed)
660 } else {
661 Some(ResolvedIntrinsicTransition::Legacy(outcome))
662 }
663 });
664 if let Some(resolved) = resolved {
665 if resolved.distribute() && resolved.can_emit() {
666 out.push((oid, resolved));
667 }
668 }
669 }
670 out
671 };
672 for (oid, resolved) in fired {
673 Self::build_and_send_event_notification_with_bindings(
674 &db_intrinsic,
675 &network_intrinsic,
676 &comm_state_intrinsic,
677 &server_tsm_intrinsic,
678 ¬ification_transactions_intrinsic,
679 &device_bindings_intrinsic,
680 &oid,
681 resolved,
682 intrinsic_retry_ms,
683 )
684 .await;
685 }
686 }
687 }));
688
689 let binary_lighting_operation_task = Some(
690 super::binary_lighting_lifecycle::spawn_binary_lighting_operation_task(
691 Arc::clone(&db),
692 Arc::clone(&network),
693 Arc::clone(&cov_table),
694 Arc::clone(&cov_in_flight),
695 Arc::clone(¬ification_transactions),
696 Arc::clone(&comm_state),
697 config.clone(),
698 monotonic_origin,
699 ),
700 );
701
702 let server = Self {
703 config,
704 _clock: clock,
705 network,
706 db,
707 cov_table,
708 seg_ack_senders,
709 seg_send_permits,
710 cov_in_flight,
711 server_tsm,
712 notification_transactions,
713 confirmed_request_tracker,
714 device_bindings,
715 comm_state,
716 dcc_timer,
717 dispatch_task: Some(dispatch_task),
718 cov_purge_task: Some(cov_purge_task),
719 fault_detection_task,
720 event_enrollment_task,
721 trend_log_task,
722 schedule_tick_task,
723 intrinsic_reporting_task,
724 binary_lighting_operation_task,
725 local_mac,
726 };
727 let staging_oids = {
728 let database = server.db.read().await;
729 database.find_by_type(ObjectType::STAGING)
730 };
731 let staging_plans = {
732 let mut database = server.db.write().await;
733 Self::take_staging_plans(&mut database, &staging_oids)
734 };
735 Self::execute_staging_plans(
736 &server.db,
737 &server.network,
738 &server.cov_table,
739 &server.cov_in_flight,
740 &server.server_tsm,
741 &server.notification_transactions,
742 &server.device_bindings,
743 &server.comm_state,
744 &server.config,
745 staging_plans,
746 )
747 .await;
748 Ok(server)
749 }
750
751 /// Send a `'server' = TRUE` Abort back along the request's path.
752 ///
753 /// Every Abort this dispatch loop originates answers a client's request,
754 /// so the flag is always TRUE — it names the sender's role, not the
755 /// error (Clause 20.1.9.1: "TRUE when the Abort PDU is sent by a
756 /// server").
757 async fn send_server_abort(
758 network: &Arc<NetworkLayer<T>>,
759 source_mac: &MacAddr,
760 source_network: Option<&NpduAddress>,
761 invoke_id: u8,
762 abort_reason: AbortReason,
763 ) {
764 let abort_pdu = Apdu::Abort(AbortPdu {
765 sent_by_server: true,
766 invoke_id,
767 abort_reason,
768 });
769 let mut abort_buf = BytesMut::new();
770 encode_apdu(&mut abort_buf, &abort_pdu).expect("valid APDU encoding");
771 if let Err(e) =
772 Self::send_confirmed_response_apdu(network, &abort_buf, source_mac, source_network)
773 .await
774 {
775 warn!(error = %e, reason = abort_reason.to_raw(), "Failed to send Abort");
776 }
777 }
778
779 /// Get the server's local MAC address.
780 pub fn local_mac(&self) -> &[u8] {
781 &self.local_mac
782 }
783
784 /// Get a reference to the shared object database.
785 pub fn database(&self) -> &Arc<RwLock<ObjectDatabase>> {
786 &self.db
787 }
788
789 /// Create a cloneable handle for unsolicited I-Am announcements.
790 pub fn i_am_broadcaster(&self) -> IAmBroadcaster<T> {
791 IAmBroadcaster {
792 config: self.config.clone(),
793 network: Arc::clone(&self.network),
794 db: Arc::clone(&self.db),
795 }
796 }
797
798 /// Get the communication state per DeviceCommunicationControl.
799 ///
800 /// Returns 0 (Enable), 1 (Disable), or 2 (DisableInitiation).
801 pub fn comm_state(&self) -> u8 {
802 self.comm_state.load(Ordering::Acquire)
803 }
804
805 /// Generate a PICS document from the current object database and server configuration.
806 ///
807 /// The caller must supply a [`PicsConfig`] for fields not available from the server
808 /// (vendor name, model, firmware revision, etc.).
809 pub async fn generate_pics(&self, pics_config: &crate::pics::PicsConfig) -> crate::pics::Pics {
810 let db = self.db.read().await;
811 crate::pics::PicsGenerator::new(&db, &self.config, pics_config).generate()
812 }
813
814 /// Broadcast an I-Am for this server's Device object using the bound transport socket.
815 pub async fn broadcast_i_am(&self) -> Result<(), Error> {
816 broadcast_i_am_from(&self.config, &self.db, &self.network).await
817 }
818}
819
820impl<T: TransportPort + 'static> IAmBroadcaster<T> {
821 /// Broadcast an I-Am for this server's Device object using the bound transport socket.
822 pub async fn broadcast_i_am(&self) -> Result<(), Error> {
823 broadcast_i_am_from(&self.config, &self.db, &self.network).await
824 }
825}
826
827async fn broadcast_i_am_from<T: TransportPort + 'static>(
828 config: &ServerConfig,
829 db: &Arc<RwLock<ObjectDatabase>>,
830 network: &Arc<NetworkLayer<T>>,
831) -> Result<(), Error> {
832 let device_oid = {
833 let db = db.read().await;
834 db.list_objects()
835 .into_iter()
836 .find(|oid| oid.object_type() == ObjectType::DEVICE)
837 .ok_or_else(|| Error::Encoding("no Device object in database".into()))?
838 };
839
840 let i_am = IAmRequest {
841 object_identifier: device_oid,
842 max_apdu_length: config.max_apdu_length,
843 segmentation_supported: config.segmentation_supported,
844 vendor_id: config.vendor_id,
845 };
846
847 let mut service_buf = BytesMut::new();
848 i_am.encode(&mut service_buf);
849
850 let pdu = Apdu::UnconfirmedRequest(UnconfirmedRequestPdu {
851 service_choice: UnconfirmedServiceChoice::I_AM,
852 service_request: service_buf.freeze(),
853 });
854
855 let mut buf = BytesMut::new();
856 encode_apdu(&mut buf, &pdu)?;
857
858 network
859 .broadcast_apdu(&buf, false, NetworkPriority::NORMAL)
860 .await
861}