bacnet_server/server/lifecycle.rs
1use super::*;
2#[path = "intrinsic_reporting_lifecycle.rs"]
3mod intrinsic;
4use crate::committed_cov::BackgroundCommit;
5
6#[path = "lifecycle_period.rs"]
7mod period;
8use super::heap_futures::boxed;
9use super::{audit_recipient::spawn_owned, audit_recipient_routes::AuditRoutes};
10pub(super) use period::{MonotonicClocks, event_enrollment_period};
11
12impl<T: TransportPort + 'static> BACnetServer<T> {
13 /// Start a server: every public start and build path ends here. The
14 /// startup future is on the heap, which keeps theirs small (#953).
15 pub(super) fn start_with_clock_mode_and_bindings(
16 config: ServerConfig,
17 db: ObjectDatabase,
18 transport: T,
19 clock_config: Option<ClockConfig>,
20 configured_device_bindings: Vec<DeviceBinding>,
21 ) -> std::pin::Pin<Box<impl std::future::Future<Output = Result<Self, Error>>>> {
22 boxed(|| {
23 Self::start_on_heap(
24 config,
25 db,
26 transport,
27 clock_config,
28 configured_device_bindings,
29 )
30 })
31 }
32
33 async fn start_on_heap(
34 mut config: ServerConfig,
35 mut db: ObjectDatabase,
36 transport: T,
37 clock_config: Option<ClockConfig>,
38 configured_device_bindings: Vec<DeviceBinding>,
39 ) -> Result<Self, Error> {
40 if let Some(profile) = &mut config.audit_reporters {
41 profile.canonicalize()?;
42 }
43 // Validate every configured route against the concrete transport before
44 // mutating the database or starting network work. No binding takes a
45 // group address (#1493).
46 // Max_APDU_Length_Accepted is fixed here for the server's life; a
47 // later fall in the transport's capacity leaves it as is, and each
48 // reply is sized to its route instead (#2093).
49 super::network_port::validate_apdu_capacity(&mut config, &transport)?;
50 crate::local_device::validate_startup(&db, &config.device)?;
51 let is_group = |mac: &[u8]| transport.is_group_destination(mac);
52 let device_bindings =
53 DeviceBindingTable::from_configured(configured_device_bindings, is_group)?;
54 let audit_routes = AuditRoutes::prepare(&mut db, &config, &device_bindings, &transport)?;
55 super::audit_forwarder::initialize(&db, &config, &device_bindings, &transport);
56 let receive_timeout = super::segmented_receive::validate_segment_timeout(&config)?;
57 // Settled now: every owner from here on shares it (#1521).
58 let config = Arc::new(config);
59 let request_tasks = super::request_tasks::RequestTasks::for_server(&config)?;
60
61 if config.device.vendor_id() == 0 {
62 warn!("vendor_id is 0 (ASHRAE reserved); set a valid vendor ID for production use");
63 }
64
65 let (clock, monotonic) = period::install_database_clocks(&mut db, clock_config);
66 // The device is starting: colour outputs go to their defaults
67 // (#1526). The database isn't shared yet, so no lock is taken. This
68 // runs before `network_port::start`, which can still fail; that's
69 // harmless, since the drive is what any start does, and a later
70 // start over the same objects drives them the same way again.
71 db.for_each_object_mut(|_, object| object.start_output_internal());
72 let membership = crate::membership::install_waker(&mut db);
73
74 let (network, mut apdu_rx, audit_routes, network_controls) =
75 boxed(|| super::network_port::start(&mut db, &config, transport, audit_routes)).await?;
76 let local_mac = MacAddr::from_slice(network.local_mac());
77
78 let network = Arc::new(network);
79 // The limiter keeps this startup identity; changing Device membership
80 // later does not rebind it.
81 let device_instance = Some(config.device.instance());
82 let (discovery_limiter, time_sync_limiter) = request_limiters(&config, device_instance);
83 let db = Arc::new(RwLock::new(db));
84 let cov_counters = Arc::new(crate::cov::AtomicCovCounters::default());
85 let cov_table = Arc::new(RwLock::new(
86 CovSubscriptionTable::with_policy(config.cov_policy.clone(), Arc::clone(&cov_counters))
87 .with_max_apdu_length(u32::from(config.device.max_apdu_length()) as usize),
88 ));
89 let seg_ack_senders = Arc::new(segmented_send::SegmentedSendRegistry::default());
90 let seg_send_permits = Arc::new(Semaphore::new(MAX_SEG_SENDERS));
91
92 let cov_in_flight = Arc::new(Semaphore::new(255));
93 let learned_routers = Arc::new(Mutex::new(LearnedRouterCache::new()));
94 let notification_transactions = NotificationTransactions::new();
95 let confirmed_request_tracker = Arc::new(ConfirmedRequestTracker::default());
96 let device_bindings = Arc::new(RwLock::new(device_bindings));
97 let comm_state = Arc::new(CommState::default()); // ENABLE at every start
98 let dcc_timer: Arc<Mutex<crate::server::dcc_timer::TimerSlot>> =
99 Arc::new(Mutex::new(Default::default()));
100 let dcc_outcomes = Arc::new(dcc_outcomes::DccOutcomes::default());
101 let event_suppressions = Arc::new(super::event_suppression::EventSuppressions::default());
102 let mutation_decisions = Arc::new(crate::mutation::MutationDecisions::default());
103
104 let target_audit = super::audit_recipient::TargetAudit::install(
105 &mut *db.write().await,
106 &config,
107 audit_routes,
108 &network,
109 ¬ification_transactions,
110 );
111 let target_audit = match target_audit {
112 Ok(runtime) => runtime,
113 Err(error) => {
114 let network = Arc::try_unwrap(network).map_err(|_| {
115 Error::Encoding("startup cleanup has an unexpected network owner".into())
116 })?;
117 boxed(|| super::network_port::StartingNetwork::from_network(network).cleanup())
118 .await?;
119 return Err(error);
120 }
121 };
122 let network_number_task = network_controls.map(|controls| {
123 super::network_port::spawn_number_worker(
124 &network,
125 &db,
126 config.registered_network_port,
127 controls,
128 target_audit.as_ref().map(Arc::downgrade),
129 )
130 });
131 let network_dispatch = Arc::clone(&network);
132 let config_dispatch = Arc::clone(&config);
133 let notification_transactions_dispatch = Arc::clone(¬ification_transactions);
134
135 let audit_owner = target_audit
136 .as_ref()
137 .map(|runtime| Arc::clone(&runtime.owner));
138 if let Some(owner) = &audit_owner {
139 request_tasks.set_audit_owner(owner);
140 }
141 let requests = Arc::clone(&request_tasks);
142 let dispatch_context = DispatchContext {
143 services: Arc::new(RequestServices {
144 db: Arc::clone(&db),
145 network: Arc::clone(&network_dispatch),
146 cov_table: Arc::clone(&cov_table),
147 seg_ack_senders: Arc::clone(&seg_ack_senders),
148 seg_send_permits: Arc::clone(&seg_send_permits),
149 cov_in_flight: Arc::clone(&cov_in_flight),
150 learned_routers: Arc::clone(&learned_routers),
151 notification_transactions: Arc::clone(¬ification_transactions_dispatch),
152 device_bindings: Arc::clone(&device_bindings),
153 comm_state: Arc::clone(&comm_state),
154 dcc_timer: Arc::clone(&dcc_timer),
155 dcc_outcomes: Arc::clone(&dcc_outcomes),
156 event_suppressions: Arc::clone(&event_suppressions),
157 confirmed_event_repeats: Arc::default(),
158 received_event_log: Arc::default(),
159 mutation_decisions: Arc::clone(&mutation_decisions),
160 warnings: Arc::default(),
161 config: Arc::clone(&config_dispatch),
162 }),
163 confirmed_request_tracker: Arc::clone(&confirmed_request_tracker),
164 clock: clock.clone(),
165 discovery_limiter: discovery_limiter.clone(),
166 time_sync_limiter: time_sync_limiter.clone(),
167 request_tasks: Arc::clone(&request_tasks),
168 };
169 // A Schedule tick that writes a Command's Present_Value starts its list.
170 let schedule_runner = super::command_runs::CommandRunner::new(
171 &dispatch_context.services,
172 &request_tasks.spawner(),
173 );
174 let dispatch_task = spawn_owned(audit_owner.clone(), move || async move {
175 let mut seg_receivers: HashMap<SegRecvKey, SegmentedRequestState> = HashMap::new();
176 let mut notifications_open = true;
177 let mut ingress_open = true;
178 // A peer's flood warns once per interval, the rest at debug (#1950).
179 let warnings = Arc::clone(&dispatch_context.services.warnings);
180 let (decode_warning, segment_warning) = (&warnings.decode, &warnings.segment);
181
182 // While ingress is open, finished handlers are joined at the top
183 // of each turn, not awaited below (IdleReap, #1713).
184 let mut idle_reap = super::request_tasks::IdleReap::new();
185 loop {
186 requests.reap_finished();
187 // Drain all selected ownership before the first asynchronous Abort.
188 Self::reap_expired_requests(
189 &network_dispatch,
190 &mut seg_receivers,
191 runtime_clock::now(),
192 receive_timeout,
193 )
194 .await;
195 let received = tokio::select! {
196 () = super::segmented_receive::wait_receive_deadline(&seg_receivers, receive_timeout), if !seg_receivers.is_empty() => continue,
197 result = notification_transactions_dispatch.join_next(), if notifications_open => {
198 notifications_open = result.is_some();
199 NotificationTransactions::observe(result);
200 continue;
201 }
202 result = requests.join_next(), if !ingress_open && !requests.is_empty() => {
203 super::request_tasks::RequestTasks::observe(result);
204 continue;
205 }
206 () = idle_reap.tick(), if ingress_open && !requests.is_empty() => continue,
207 received = apdu_rx.recv(), if ingress_open => match received {
208 Some(received) => received,
209 None => {
210 // Local/periodic notification producers outlive ingress.
211 // Keep both join consumers active, but never poll EOF again.
212 ingress_open = false;
213 seg_receivers.clear();
214 continue;
215 }
216 },
217 else => break,
218 };
219 // Input and the timer can become ready together. Apply expiry
220 // before interpreting this segment in either scheduling order.
221 Self::reap_expired_requests(
222 &network_dispatch,
223 &mut seg_receivers,
224 runtime_clock::now(),
225 receive_timeout,
226 )
227 .await;
228
229 match apdu::decode_apdu(received.apdu.clone()) {
230 Ok(decoded) => {
231 if notification_transactions_dispatch
232 .application_sealed
233 .load(Ordering::Acquire)
234 {
235 seg_receivers.clear();
236 if matches!(
237 decoded,
238 Apdu::ConfirmedRequest(_) | Apdu::UnconfirmedRequest(_)
239 ) {
240 continue;
241 }
242 }
243 // A confirmed request may only be addressed to one
244 // device (Clause 6.3). One that arrives by local,
245 // remote or global broadcast, or by multicast, leaves
246 // the server TSM idle (Clause 5.4.5.1): it is neither
247 // executed nor answered, whatever the service, and
248 // no reassembly starts for it.
249 if received.is_group && matches!(decoded, Apdu::ConfirmedRequest(_)) {
250 debug!("Ignoring a ConfirmedRequest addressed to a group");
251 continue;
252 }
253 // Nor one sent from a group address, where its
254 // answer would go (#1504).
255 if matches!(decoded, Apdu::ConfirmedRequest(_))
256 && dispatch_context
257 .confirmed_request_tracker
258 .refuse_group_source(&received.source_mac, |mac| {
259 network_dispatch.transport().is_group_destination(mac)
260 })
261 {
262 continue;
263 }
264 let source_mac = received.source_mac.clone();
265 let source_network = received.source_network.clone();
266
267 // Clause 5.4.5.2 AbortPDU_Received: a peer's Abort
268 // ('server' = FALSE) ends any reassembly session for
269 // its transaction. A side effect, not a short
270 // circuit — the PDU still reaches `dispatch`, whose
271 // Abort arm cancels in-flight segmented response
272 // senders and records server-TSM results (#377).
273 // Provenance snapshot for this ingress (RB-07, by value).
274 let provenance = received.provenance.clone();
275 let route = received.response_route();
276 super::segmented_receive::remove_peer_aborted_request(
277 &mut seg_receivers,
278 &received,
279 &decoded,
280 );
281
282 let mut received = Some(received);
283 let handled = if let Apdu::ConfirmedRequest(ref req) = decoded {
284 if req.segmented {
285 let seq = req.sequence_number.unwrap_or(0);
286 let key = segmented_receive_key(
287 source_mac.as_slice(),
288 source_network.as_ref(),
289 req.invoke_id,
290 provenance.clone(),
291 );
292 if let Some(conflict) =
293 super::segmented_receive::find_receive_provenance_conflict(
294 &seg_receivers,
295 &key,
296 )
297 {
298 seg_receivers.remove(&conflict);
299 warn_throttled!(
300 segment_warning,
301 invoke_id = req.invoke_id,
302 "Aborting segmented request on provenance mismatch (fail-closed)"
303 );
304 Self::send_server_abort(
305 &network_dispatch,
306 &source_mac,
307 source_network.as_ref(),
308 &route,
309 req.invoke_id,
310 AbortReason::INVALID_APDU_IN_THIS_STATE,
311 )
312 .await;
313 continue;
314 }
315
316 if !Self::accept_segment_route(
317 &network_dispatch,
318 &mut seg_receivers,
319 &key,
320 &route,
321 )
322 .await
323 {
324 continue;
325 }
326
327 // Clause 5.4.5.1
328 // ConfirmedSegmentedReceivedNotSupported: a
329 // device that does not support segmented
330 // reception answers segment traffic with this
331 // Abort instead of reassembling — the
332 // configured Segmentation value is the
333 // advertisement peers plan transfers around
334 // (#381).
335 let receives_segments = config_dispatch.device.segmentation()
336 == Segmentation::BOTH
337 || config_dispatch.device.segmentation()
338 == Segmentation::RECEIVE;
339 if !receives_segments {
340 Self::send_server_abort(
341 &network_dispatch,
342 &source_mac,
343 source_network.as_ref(),
344 &route,
345 req.invoke_id,
346 AbortReason::SEGMENTATION_NOT_SUPPORTED,
347 )
348 .await;
349 continue;
350 }
351
352 let mut ack_to_send: Option<SegmentAckPdu> = None;
353 let mut final_total: Option<usize> = None;
354
355 // The live session is consulted before the
356 // `seq == 0` open path: Clause 20.1.2.7 wraps
357 // the sequence number modulo 256, so segment
358 // 256 of a long request arrives as another
359 // `seq == 0` — treating it as a fresh initial
360 // segment would silently replace the session
361 // and reassemble only the tail (#364).
362 let saved_bytes =
363 super::segmented_receive::saved_request_payload_bytes(
364 &seg_receivers,
365 );
366 if let Some(state) = seg_receivers.get_mut(&key) {
367 // Compat-mode live read: key isolates
368 // contexts; snapshot must match the key.
369 debug_assert_eq!(state.provenance, provenance);
370 // Clause 5.4.5.2 restarts SegmentTimer
371 // for accepted, duplicate and
372 // out-of-order segments alike, so the
373 // refresh precedes the ordering checks.
374 state.last_activity = runtime_clock::now();
375 if seq != state.expected_seq {
376 ack_to_send =
377 super::segmented_receive::classify_non_next_segment(
378 state,
379 req.invoke_id,
380 seq,
381 segment_warning,
382 );
383 } else {
384 // In-order NEW segment: duplicates
385 // and gaps returned above, so a
386 // retransmission can never trip the
387 // cap (Clause 5.4.5.2
388 // DuplicateSegmentReceived requires
389 // duplicates be discarded, not
390 // punished).
391 if state.accepted_segments >= MAX_REQUEST_SEGMENTS {
392 // Clause 5.4.5.2 has no overflow
393 // transition; SendAbort ('server'
394 // = TRUE, reason a local matter)
395 // is its one generic escape, and
396 // Clause 18.10's BUFFER_OVERFLOW
397 // fits a reassembly that exceeds
398 // available buffer capacity (#364).
399 warn_throttled!(
400 segment_warning,
401 invoke_id = req.invoke_id,
402 accepted = state.accepted_segments,
403 "Segmented request exceeds reassembly capacity, aborting"
404 );
405 seg_receivers.remove(&key);
406 Self::send_server_abort(
407 &network_dispatch,
408 &source_mac,
409 source_network.as_ref(),
410 &route,
411 req.invoke_id,
412 AbortReason::BUFFER_OVERFLOW,
413 )
414 .await;
415 continue;
416 }
417 if let Err(e) = state.payload.save_new(
418 seq,
419 req.service_request.clone(),
420 saved_bytes,
421 ) {
422 // An unsaveable segment ends the
423 // session the same way — leaving
424 // it dangling told the peer
425 // nothing while this side could
426 // never complete (#364).
427 warn_throttled!(
428 segment_warning,
429 error = %e,
430 "Rejecting unsaveable segment"
431 );
432 seg_receivers.remove(&key);
433 Self::send_server_abort(
434 &network_dispatch,
435 &source_mac,
436 source_network.as_ref(),
437 &route,
438 req.invoke_id,
439 AbortReason::BUFFER_OVERFLOW,
440 )
441 .await;
442 continue;
443 }
444 state.last_progress = runtime_clock::now();
445 state.accepted_segments += 1;
446 state.expected_seq = seq.wrapping_add(1);
447 state.last_acked_seq = seq;
448 state.window_pos += 1;
449 let should_ack = !req.more_follows
450 || state.window_pos >= state.actual_window_size;
451 if should_ack {
452 state.window_pos = 0;
453 state.initial_sequence_number = state.last_acked_seq;
454 state.duplicate_count = 0;
455 ack_to_send = Some(SegmentAckPdu {
456 negative_ack: false,
457 sent_by_server: true,
458 invoke_id: req.invoke_id,
459 sequence_number: seq,
460 actual_window_size: state.actual_window_size,
461 });
462 }
463 if !req.more_follows {
464 // The count, not `seq + 1`: the
465 // wire sequence number is modulo
466 // 256 (Clause 20.1.2.7) and says
467 // nothing about how many segments
468 // were accepted (#364).
469 final_total = Some(state.accepted_segments);
470 }
471 }
472 } else if seq == 0 {
473 let proposed_window_size =
474 req.proposed_window_size.unwrap_or(0);
475 if !(1..=127).contains(&proposed_window_size) {
476 warn_throttled!(
477 segment_warning,
478 invoke_id = req.invoke_id,
479 proposed_window_size,
480 "Rejecting segmented request with invalid proposed window size"
481 );
482 Self::send_server_abort(
483 &network_dispatch,
484 &source_mac,
485 source_network.as_ref(),
486 &route,
487 req.invoke_id,
488 AbortReason::WINDOW_SIZE_OUT_OF_RANGE,
489 )
490 .await;
491 continue;
492 }
493
494 // New sessions only; global capacity precedes peer quota.
495 if let Some(reason) =
496 super::segmented_receive::segmented_request_admission_error(
497 &seg_receivers,
498 &key,
499 )
500 {
501 Self::send_server_abort(
502 &network_dispatch,
503 &source_mac,
504 source_network.as_ref(),
505 &route,
506 req.invoke_id,
507 reason,
508 )
509 .await;
510 continue;
511 }
512
513 let mut payload =
514 super::segmented_receive::RequestPayload::new(req);
515 if let Err(e) = payload.save_new(
516 seq,
517 req.service_request.clone(),
518 saved_bytes,
519 ) {
520 // No session exists to drop on this
521 // path; the Abort is what tells the
522 // peer instead of leaving it to time
523 // out (#364).
524 warn_throttled!(
525 segment_warning,
526 error = %e,
527 "Rejecting unsaveable segment"
528 );
529 Self::send_server_abort(
530 &network_dispatch,
531 &source_mac,
532 source_network.as_ref(),
533 &route,
534 req.invoke_id,
535 AbortReason::BUFFER_OVERFLOW,
536 )
537 .await;
538 continue;
539 }
540 let (state, initial_ack) =
541 super::segmented_receive::initial_state(
542 payload,
543 route.clone(),
544 source_mac.clone(),
545 source_network.clone(),
546 req,
547 );
548 ack_to_send = initial_ack;
549 if !req.more_follows {
550 final_total = Some(1);
551 }
552 seg_receivers.insert(key.clone(), state);
553 } else {
554 warn_throttled!(
555 segment_warning,
556 invoke_id = req.invoke_id,
557 seq = seq,
558 "Received non-initial segment without prior segment 0, aborting"
559 );
560 Self::send_server_abort(
561 &network_dispatch,
562 &source_mac,
563 source_network.as_ref(),
564 &route,
565 req.invoke_id,
566 AbortReason::INVALID_APDU_IN_THIS_STATE,
567 )
568 .await;
569 continue;
570 }
571
572 if let Some(seg_ack) = ack_to_send {
573 let seg_ack = Apdu::SegmentAck(seg_ack);
574 let mut ack_buf = BytesMut::new();
575 encode_apdu(&mut ack_buf, &seg_ack)
576 .expect("valid APDU encoding");
577 if let Err(e) = Self::send_confirmed_response_apdu(
578 &network_dispatch,
579 &ack_buf,
580 &source_mac,
581 source_network.as_ref(),
582 &route,
583 )
584 .await
585 {
586 warn!(
587 error = %e,
588 "Failed to send SegmentAck for segmented request"
589 );
590 }
591 }
592
593 if let Some(total) = final_total
594 && let Some(state) = seg_receivers.remove(&key)
595 {
596 if let Some(envelope) = received.as_mut() {
597 envelope.set_response_route(state.route.clone());
598 }
599 match state.payload.complete(total) {
600 Ok(reassembled) => {
601 debug!(
602 invoke_id = reassembled.invoke_id,
603 segments = total,
604 payload_len = reassembled.service_request.len(),
605 "Reassembled segmented ConfirmedRequest"
606 );
607 Self::dispatch(
608 &dispatch_context,
609 &source_mac,
610 Apdu::ConfirmedRequest(reassembled),
611 received.take().unwrap_or_else(
612 super::segmented_receive::empty_reassembled_envelope,
613 ),
614 )
615 .await;
616 }
617 Err(e) => {
618 warn_throttled!(
619 segment_warning,
620 error = %e,
621 "Failed to reassemble segmented request"
622 );
623 }
624 }
625 }
626
627 true
628 } else {
629 false
630 }
631 } else {
632 false
633 };
634
635 if !handled {
636 Self::dispatch(
637 &dispatch_context,
638 &source_mac,
639 decoded,
640 received.take().unwrap_or_else(
641 super::segmented_receive::empty_reassembled_envelope,
642 ),
643 )
644 .await;
645 }
646 }
647 Err(e) => {
648 warn_throttled!(
649 decode_warning,
650 error = %e,
651 "Server failed to decode received APDU"
652 );
653 }
654 }
655 }
656 });
657
658 let cov_table_for_purge = Arc::clone(&cov_table);
659 let cov_purge_task = spawn_owned(audit_owner.clone(), move || async move {
660 let mut interval = tokio::time::interval(Duration::from_secs(30));
661 loop {
662 interval.tick().await;
663 let mut table = cov_table_for_purge.write().await;
664 let purged = table.purge_expired();
665 if purged > 0 {
666 debug!(purged, "Purged expired COV subscriptions");
667 }
668 }
669 });
670
671 // Background commits fan COV out like a network write (#889).
672 let cov_fanout = super::cov_fanout::CovFanout::new(
673 &super::cov_notify_context::CovNotifyContext {
674 db: &db,
675 network: &network,
676 cov_table: &cov_table,
677 cov_in_flight: &cov_in_flight,
678 notification_transactions: ¬ification_transactions,
679 comm_state: &comm_state,
680 config: &config,
681 },
682 &event_suppressions,
683 );
684
685 let fault_detection_task = if config.enable_fault_detection {
686 let fanout = cov_fanout.clone();
687 Some(spawn_owned(audit_owner.clone(), move || async move {
688 let detector = crate::fault_detection::FaultDetector::default();
689 let mut interval = tokio::time::interval(Duration::from_secs(10));
690 loop {
691 interval.tick().await;
692 let committed = {
693 let mut db_guard = fanout.db.write().await;
694 // The detector mutates objects in place, so snapshot
695 // Life Safety state up front.
696 let mut commit = BackgroundCommit::snapshot_all(&db_guard);
697 for change in detector.evaluate(&mut db_guard) {
698 debug!(
699 object = %change.object_id,
700 old = %change.old_reliability,
701 new = %change.new_reliability,
702 "Fault detection: reliability changed"
703 );
704 commit.changed(change.object_id);
705 }
706 commit
707 .finish(&fanout.db, &mut db_guard, &fanout.cov_table)
708 .await
709 };
710 fanout.fire(&committed).await;
711 }
712 }))
713 } else {
714 None
715 };
716
717 let event_enrollment_task = if config.enable_event_enrollment {
718 let ee_period = event_enrollment_period(config.event_enrollment_interval_secs);
719 // The delay countdown converts seconds to passes with
720 // `ceil(delay / period)`, so the evaluator needs the actual,
721 // clamped interval — not the raw config value.
722 Some(
723 super::event_enrollment_lifecycle::spawn_event_enrollment_task(
724 super::event_enrollment_lifecycle::EventEnrollmentTask {
725 db: Arc::clone(&db),
726 network: Arc::clone(&network),
727 comm_state: Arc::clone(&comm_state),
728 learned_routers: Arc::clone(&learned_routers),
729 notification_transactions: Arc::clone(¬ification_transactions),
730 device_bindings: Arc::clone(&device_bindings),
731 suppressions: Arc::clone(&event_suppressions),
732 period: ee_period,
733 retry_ms: config.cov_retry_timeout_ms,
734 local_apdu_capacity: u32::from(config.device.max_apdu_length()),
735 },
736 ),
737 )
738 } else {
739 None
740 };
741
742 let trend_log_task = Some(spawn_owned(audit_owner.clone(), || {
743 crate::trend_log::run(Arc::clone(&db))
744 }));
745
746 let schedule_fanout = cov_fanout.clone();
747 // The 60-second Schedule pass, and the work an application's own
748 // `add` or `remove` queued on the database (`crate::membership`).
749 let schedule_tick_task = Some(spawn_owned(audit_owner.clone(), move || async move {
750 let mut interval = tokio::time::interval(Duration::from_secs(60));
751 loop {
752 let (db, cov_table) = (&schedule_fanout.db, &schedule_fanout.cov_table);
753 let committed = tokio::select! {
754 _ = interval.tick() => {
755 crate::schedule::tick_schedules_committed(db, cov_table).await
756 }
757 () = membership.notified() => {
758 crate::membership::settle_committed(db, cov_table).await
759 }
760 };
761 schedule_fanout.fire(&committed).await;
762 schedule_runner.start(committed.command_runs);
763 }
764 }));
765
766 // One-second intrinsic-reporting task: advances the `Time_Delay`
767 // countdown for any object with a pending delayed transition and sends
768 // the EventNotification when the delay elapses. The per-write path
769 // only *seeds* a pending transition (see `fire_event_notifications`);
770 // this task is the sole confirmer, so repeated writes cannot shorten
771 // the delay (ASHRAE 135-2020 Clause 13.3). Runs unconditionally like the
772 // trend-log task — a no-pending tick is a cheap empty iteration.
773 //
774 // It is also what carries Reliability into event-state-detection. Per
775 // Clause 13.2.2 the FAULT determination is a standing condition, so each
776 // tick re-derives it from the object's current `Reliability` rather than
777 // reacting to a change event. Whoever writes Reliability — an object's
778 // opt-in evaluation hook, a local write, or a network write — reaches
779 // detection through this tick. The fault detector above only fans COV
780 // out for the Status_Flags change it causes (#889); it signals nothing
781 // to event detection. `enable_fault_detection` therefore governs only
782 // whether those object-owned hooks run every 10 seconds, never whether
783 // an existing Reliability is honored.
784 //
785 // Six of the nine wired object types have no route that can set
786 // Reliability, so the fault path is correct but inert on them (#218).
787 let intrinsic_reporting_task = Some(spawn_owned(audit_owner.clone(), || {
788 intrinsic::run(
789 cov_fanout.clone(),
790 Arc::clone(&learned_routers),
791 Arc::clone(&device_bindings),
792 )
793 }));
794
795 // Reports what changed while a confirmed report was outstanding, once
796 // it is acknowledged (#896).
797 let cov_revisit_task = Some(spawn_owned(audit_owner.clone(), || {
798 cov_fanout.clone().run_revisits()
799 }));
800
801 let binary_lighting_operation_task = Some(
802 super::binary_lighting_lifecycle::spawn_binary_lighting_operation_task(
803 cov_fanout, monotonic,
804 ),
805 );
806
807 let broadcaster = super::broadcaster::BroadcasterState::new(
808 &network,
809 &request_tasks,
810 &config,
811 &db,
812 &comm_state,
813 );
814 let server = Self {
815 target_audit,
816 config,
817 discovery_limiter,
818 time_sync_limiter,
819 _clock: clock,
820 network: Some(network),
821 broadcaster,
822 transport_cleanup: None,
823 transport_cleanup_error: None,
824 network_number_task,
825 db,
826 cov_table,
827 cov_counters,
828 seg_ack_senders,
829 seg_send_permits,
830 cov_in_flight,
831 learned_routers,
832 notification_transactions,
833 confirmed_request_tracker,
834 device_bindings,
835 comm_state,
836 dcc_timer,
837 dcc_outcomes,
838 event_suppressions,
839 mutation_decisions,
840 dispatch_task: Some(dispatch_task),
841 request_tasks,
842 cov_purge_task: Some(cov_purge_task),
843 fault_detection_task,
844 event_enrollment_task,
845 trend_log_task,
846 schedule_tick_task,
847 intrinsic_reporting_task,
848 binary_lighting_operation_task,
849 cov_revisit_task,
850 local_mac,
851 };
852 boxed(|| server.execute_initial_staging_plans()).await;
853 Ok(server)
854 }
855}