1use std::collections::{BTreeMap, HashSet};
2use std::fs;
3use std::path::{Path, PathBuf};
4
5use serde::{Deserialize, Serialize};
6use tempfile::NamedTempFile;
7
8use crate::config::{
9 ConfigBase, EntityConfig, IncrementalMode, ResolvedPath, RootConfig, StorageResolver,
10};
11use crate::io::storage::{
12 extensions, local::LocalClient, CloudClient, ConditionalWrite, StorageClient, StoredObject,
13};
14use crate::{ConfigError, FloeResult};
15
16pub const ENTITY_STATE_SCHEMA_V1: &str = "floe.state.file-ingest.v1";
17pub const ENTITY_STATE_SCHEMA_V2: &str = "floe.state.file-ingest.v2";
18pub const ENTITY_STATE_FILENAME: &str = "state.json";
19const STATE_CAS_RETRIES: usize = 5;
20pub const CLAIM_TTL_SECONDS: i64 = 60 * 60;
21
22#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
23pub struct EntityState {
24 pub schema: String,
25 pub entity: String,
26 pub updated_at: Option<String>,
27 #[serde(default)]
28 pub files: BTreeMap<String, EntityFileState>,
29 #[serde(default)]
30 pub claims: BTreeMap<String, EntityFileClaim>,
31}
32
33impl EntityState {
34 pub fn new(entity: impl Into<String>) -> Self {
35 Self {
36 schema: ENTITY_STATE_SCHEMA_V2.to_string(),
37 entity: entity.into(),
38 updated_at: None,
39 files: BTreeMap::new(),
40 claims: BTreeMap::new(),
41 }
42 }
43}
44
45#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
46pub struct EntityFileState {
47 pub processed_at: String,
48 pub size: Option<u64>,
49 pub mtime: Option<String>,
50}
51
52#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
53pub struct EntityFileClaim {
54 pub run_id: String,
55 pub acquired_at: String,
56 pub expires_at: String,
57 pub size: Option<u64>,
58 pub mtime: Option<String>,
59}
60
61#[derive(Debug, Clone)]
62pub struct EntityStateInspection {
63 pub entity_name: String,
64 pub incremental_mode: IncrementalMode,
65 pub path: ResolvedPath,
66 pub state: Option<EntityState>,
67}
68
69pub fn resolve_entity_state_path(
70 resolver: &StorageResolver,
71 entity: &EntityConfig,
72) -> FloeResult<ResolvedPath> {
73 if let Some(state) = &entity.state {
74 if let Some(path) = state.path.as_deref() {
75 let resolved = if is_remote_uri(path) {
76 resolver.resolve_path(
77 &entity.name,
78 "entity.state.path",
79 entity.source.storage.as_deref(),
80 path,
81 )?
82 } else {
83 resolver.resolve_local_path(path)?
84 };
85 return Ok(resolved);
86 }
87 }
88
89 let resolved_source = resolver.resolve_path(
90 &entity.name,
91 "entity.source.path",
92 entity.source.storage.as_deref(),
93 &entity.source.path,
94 )?;
95 let source_root = derive_source_root(
96 &entity.source.path,
97 &entity.source.format,
98 resolved_source.local_path.as_deref(),
99 );
100 let default_path = join_state_path(&source_root, &entity.name);
101 let resolved = resolver.resolve_path(
102 &entity.name,
103 "entity.state.path",
104 entity.source.storage.as_deref(),
105 &default_path,
106 )?;
107 Ok(resolved)
108}
109
110pub fn read_entity_state(path: &Path) -> FloeResult<Option<EntityState>> {
111 if !path.exists() {
112 return Ok(None);
113 }
114 let payload = fs::read_to_string(path)?;
115 let state = parse_entity_state(payload.as_bytes())?;
116 Ok(Some(state))
117}
118
119fn parse_entity_state(payload: &[u8]) -> FloeResult<EntityState> {
120 let mut state: EntityState = serde_json::from_slice(payload)?;
121 if state.schema == ENTITY_STATE_SCHEMA_V1 {
122 state.schema = ENTITY_STATE_SCHEMA_V2.to_string();
123 state.claims.clear();
124 }
125 Ok(state)
126}
127
128#[derive(Debug, Clone)]
129pub enum EntityStateTarget {
130 Local { path: PathBuf, uri: String },
131 Remote { storage: String, uri: String },
132}
133
134#[derive(Debug, Clone)]
135pub struct LoadedEntityState {
136 pub target: EntityStateTarget,
137 pub state: EntityState,
138 pub version: Option<String>,
139 pub existed: bool,
140}
141
142#[derive(Debug, Clone)]
143pub struct ClaimedEntityState {
144 pub target: EntityStateTarget,
145 pub state: EntityState,
146 pub version: Option<String>,
147}
148
149#[derive(Debug, Clone)]
150pub struct EntityStateClaimOutcome {
151 pub pending_inputs: Vec<crate::io::format::InputFile>,
152 pub claimed_state: Option<ClaimedEntityState>,
153 pub active_claims: Vec<String>,
154 pub already_processed: Vec<(crate::io::format::InputFile, EntityFileState)>,
155}
156
157pub fn claim_entity_inputs(
158 resolver: &StorageResolver,
159 cloud: &mut CloudClient,
160 entity: &EntityConfig,
161 run_id: &str,
162 input_files: Vec<crate::io::format::InputFile>,
163) -> FloeResult<EntityStateClaimOutcome> {
164 if input_files.is_empty() {
165 return Ok(EntityStateClaimOutcome {
166 pending_inputs: Vec::new(),
167 claimed_state: None,
168 active_claims: Vec::new(),
169 already_processed: Vec::new(),
170 });
171 }
172
173 for _ in 0..STATE_CAS_RETRIES {
174 let mut loaded = load_entity_state(resolver, cloud, entity)?;
175 remove_expired_claims(&mut loaded.state);
176 let mut pending_inputs = Vec::new();
177 let mut active_claims = Vec::new();
178 let mut already_processed = Vec::new();
179 let acquired_at = now_rfc3339();
180 let expires_at = rfc3339_after_seconds(CLAIM_TTL_SECONDS);
181
182 for input_file in &input_files {
183 if let Some(recorded) = loaded.state.files.get(&input_file.source_uri) {
184 already_processed.push((input_file.clone(), recorded.clone()));
185 continue;
186 }
187 match loaded.state.claims.get(&input_file.source_uri) {
188 Some(_) => {
189 active_claims.push(input_file.source_uri.clone());
190 }
191 None => {
192 loaded.state.claims.insert(
193 input_file.source_uri.clone(),
194 EntityFileClaim {
195 run_id: run_id.to_string(),
196 acquired_at: acquired_at.clone(),
197 expires_at: expires_at.clone(),
198 size: input_file.source_size,
199 mtime: input_file.source_mtime.clone(),
200 },
201 );
202 pending_inputs.push(input_file.clone());
203 }
204 }
205 }
206
207 if pending_inputs.is_empty() {
208 if active_claims.is_empty() {
209 let _ = persist_loaded_state(cloud, resolver, &loaded)?;
210 }
211 return Ok(EntityStateClaimOutcome {
212 pending_inputs,
213 claimed_state: None,
214 active_claims,
215 already_processed,
216 });
217 }
218
219 loaded.state.schema = ENTITY_STATE_SCHEMA_V2.to_string();
220 loaded.state.updated_at = Some(acquired_at);
221 match persist_loaded_state(cloud, resolver, &loaded)? {
222 Some(version) => {
223 return Ok(EntityStateClaimOutcome {
224 pending_inputs,
225 claimed_state: Some(ClaimedEntityState {
226 target: loaded.target,
227 state: loaded.state,
228 version,
229 }),
230 active_claims,
231 already_processed,
232 });
233 }
234 None => continue,
235 }
236 }
237
238 Err(Box::new(ConfigError(format!(
239 "entity.name={} incremental state update conflicted after {STATE_CAS_RETRIES} retries",
240 entity.name
241 ))))
242}
243
244pub fn promote_claimed_entity_state(
245 resolver: &StorageResolver,
246 cloud: &mut CloudClient,
247 entity_name: &str,
248 run_id: &str,
249 claimed: &ClaimedEntityState,
250) -> FloeResult<()> {
251 mutate_claimed_state(resolver, cloud, entity_name, claimed, |state, our_uris| {
252 let processed_at = now_rfc3339();
253 let claimed_files: Vec<String> = state
254 .claims
255 .iter()
256 .filter(|(uri, claim)| claim.run_id == run_id && our_uris.contains(*uri))
257 .map(|(source_uri, _)| source_uri.clone())
258 .collect();
259 for source_uri in claimed_files {
260 if let Some(claim) = state.claims.remove(&source_uri) {
261 state.files.insert(
262 source_uri,
263 EntityFileState {
264 processed_at: processed_at.clone(),
265 size: claim.size,
266 mtime: claim.mtime,
267 },
268 );
269 }
270 }
271 state.updated_at = Some(processed_at);
272 })
273}
274
275pub fn release_claimed_entity_state(
276 resolver: &StorageResolver,
277 cloud: &mut CloudClient,
278 entity_name: &str,
279 run_id: &str,
280 claimed: &ClaimedEntityState,
281) -> FloeResult<()> {
282 mutate_claimed_state(resolver, cloud, entity_name, claimed, |state, our_uris| {
283 state
284 .claims
285 .retain(|uri, claim| !(claim.run_id == run_id && our_uris.contains(uri)));
286 state.updated_at = Some(now_rfc3339());
287 })
288}
289
290pub fn renew_claimed_entity_state(
291 resolver: &StorageResolver,
292 cloud: &mut CloudClient,
293 entity_name: &str,
294 run_id: &str,
295 claimed: &ClaimedEntityState,
296) -> FloeResult<()> {
297 mutate_claimed_state(resolver, cloud, entity_name, claimed, |state, our_uris| {
298 let now = now_rfc3339();
299 let expires_at = rfc3339_after_seconds(CLAIM_TTL_SECONDS);
300 for (uri, claim) in state.claims.iter_mut() {
301 if claim.run_id == run_id && our_uris.contains(uri) {
302 claim.expires_at = expires_at.clone();
303 }
304 }
305 state.updated_at = Some(now);
306 })
307}
308
309fn mutate_claimed_state(
310 resolver: &StorageResolver,
311 cloud: &mut CloudClient,
312 entity_name: &str,
313 claimed: &ClaimedEntityState,
314 mutate: impl Fn(&mut EntityState, &HashSet<String>),
315) -> FloeResult<()> {
316 let our_uris: HashSet<String> = claimed.state.claims.keys().cloned().collect();
317 for attempt in 0..STATE_CAS_RETRIES {
318 let mut loaded = if attempt == 0 {
319 LoadedEntityState {
320 target: claimed.target.clone(),
321 state: claimed.state.clone(),
322 version: claimed.version.clone(),
323 existed: claimed.version.is_some(),
324 }
325 } else {
326 load_target_state_with_entity_name(
327 cloud,
328 resolver,
329 entity_name,
330 claimed.target.clone(),
331 )?
332 };
333 mutate(&mut loaded.state, &our_uris);
334 loaded.state.schema = ENTITY_STATE_SCHEMA_V2.to_string();
335 let persisted = if loaded.state.files.is_empty() && loaded.state.claims.is_empty() {
336 delete_loaded_state(cloud, resolver, &loaded)?
337 } else {
338 persist_loaded_state(cloud, resolver, &loaded)?
339 };
340 if persisted.is_some() {
341 return Ok(());
342 }
343 }
344 Err(Box::new(ConfigError(format!(
345 "entity.name={} incremental state update conflicted after {STATE_CAS_RETRIES} retries",
346 entity_name
347 ))))
348}
349
350pub fn inspect_entity_state_with_base(
351 config_path: &Path,
352 config_base: ConfigBase,
353 entity_name: &str,
354) -> FloeResult<EntityStateInspection> {
355 let config = crate::load_config(config_path)?;
356 inspect_entity_state(&config, config_base, entity_name)
357}
358
359pub fn inspect_entity_state(
360 config: &RootConfig,
361 config_base: ConfigBase,
362 entity_name: &str,
363) -> FloeResult<EntityStateInspection> {
364 let (entity, path) = resolve_entity_state_target(config, config_base.clone(), entity_name)?;
365 let resolver = StorageResolver::new(config, config_base)?;
366 let target = state_target_from_resolved(&path)?;
367 let mut cloud = CloudClient::new();
368 let loaded = load_target_state_with_resolver(&mut cloud, &resolver, entity, target)?;
369 let state = loaded.existed.then_some(loaded.state);
370
371 Ok(EntityStateInspection {
372 entity_name: entity.name.clone(),
373 incremental_mode: entity.incremental_mode,
374 path,
375 state,
376 })
377}
378
379pub fn reset_entity_state(
380 config: &RootConfig,
381 config_base: ConfigBase,
382 entity_name: &str,
383) -> FloeResult<bool> {
384 let (entity, path) = resolve_entity_state_target(config, config_base.clone(), entity_name)?;
385 let target = state_target_from_resolved(&path)?;
386 match target {
387 EntityStateTarget::Local { path, .. } => {
388 if path.exists() {
389 fs::remove_file(&path)?;
390 return Ok(true);
391 }
392 Ok(false)
393 }
394 EntityStateTarget::Remote { storage, uri } => {
395 let mut cloud = CloudClient::new();
396 let resolver = StorageResolver::new(config, config_base)?;
397 let client = cloud.client_for_context(
398 &resolver,
399 &storage,
400 &format!("entity.name={}", entity.name),
401 )?;
402 let Some(object) = client.read_object(&uri)? else {
403 return Ok(false);
404 };
405 match client.delete_object_conditional(&uri, Some(&object.version))? {
406 ConditionalWrite::Written { .. } => Ok(true),
407 ConditionalWrite::Conflict => Err(Box::new(ConfigError(format!(
408 "entity.name={} remote state changed while resetting: {}",
409 entity.name, uri
410 )))),
411 }
412 }
413 }
414}
415
416pub fn reset_entity_state_with_base(
417 config_path: &Path,
418 config_base: ConfigBase,
419 entity_name: &str,
420) -> FloeResult<bool> {
421 let config = crate::load_config(config_path)?;
422 reset_entity_state(&config, config_base, entity_name)
423}
424
425fn resolve_entity_state_target<'a>(
426 config: &'a RootConfig,
427 config_base: ConfigBase,
428 entity_name: &str,
429) -> FloeResult<(&'a EntityConfig, ResolvedPath)> {
430 let entity = config
431 .entities
432 .iter()
433 .find(|entity| entity.name == entity_name)
434 .ok_or_else(|| {
435 Box::new(ConfigError(format!("entity not found: {entity_name}")))
436 as Box<dyn std::error::Error + Send + Sync>
437 })?;
438 let resolver = StorageResolver::new(config, config_base)?;
439 let path = resolve_entity_state_path(&resolver, entity)?;
440 Ok((entity, path))
441}
442
443pub fn write_entity_state_atomic(path: &Path, state: &EntityState) -> FloeResult<()> {
444 let parent = path.parent().ok_or_else(|| {
445 Box::new(ConfigError(format!(
446 "state path has no parent directory: {}",
447 path.display()
448 ))) as Box<dyn std::error::Error + Send + Sync>
449 })?;
450 fs::create_dir_all(parent)?;
451
452 let mut temp = NamedTempFile::new_in(parent)?;
453 serde_json::to_writer_pretty(temp.as_file_mut(), state)?;
454 temp.as_file_mut().sync_all()?;
455 temp.persist(path).map_err(|err| err.error)?;
456 Ok(())
457}
458
459fn load_entity_state(
460 resolver: &StorageResolver,
461 cloud: &mut CloudClient,
462 entity: &EntityConfig,
463) -> FloeResult<LoadedEntityState> {
464 let resolved = resolve_entity_state_path(resolver, entity)?;
465 let target = state_target_from_resolved(&resolved)?;
466 load_target_state_with_resolver(cloud, resolver, entity, target)
467}
468
469fn load_target_state_with_resolver(
470 cloud: &mut CloudClient,
471 resolver: &StorageResolver,
472 entity: &EntityConfig,
473 target: EntityStateTarget,
474) -> FloeResult<LoadedEntityState> {
475 load_target_state_with_entity_name(cloud, resolver, &entity.name, target)
476}
477
478fn load_target_state_with_entity_name(
479 cloud: &mut CloudClient,
480 resolver: &StorageResolver,
481 entity_name: &str,
482 target: EntityStateTarget,
483) -> FloeResult<LoadedEntityState> {
484 match target {
485 EntityStateTarget::Local { path, uri } => {
486 let object = LocalClient::new().read_object(&uri)?;
487 let (state, version, existed) = resolve_loaded_state(entity_name, object)?;
488 Ok(LoadedEntityState {
489 target: EntityStateTarget::Local { path, uri },
490 state,
491 version,
492 existed,
493 })
494 }
495 EntityStateTarget::Remote { storage, uri } => {
496 let client = cloud.client_for_context(
497 resolver,
498 &storage,
499 &format!("entity.name={entity_name}"),
500 )?;
501 let object = client.read_object(&uri)?;
502 let (state, version, existed) = resolve_loaded_state(entity_name, object)?;
503 Ok(LoadedEntityState {
504 target: EntityStateTarget::Remote { storage, uri },
505 state,
506 version,
507 existed,
508 })
509 }
510 }
511}
512
513fn resolve_loaded_state(
514 entity_name: &str,
515 object: Option<StoredObject>,
516) -> FloeResult<(EntityState, Option<String>, bool)> {
517 match object {
518 Some(object) => Ok((
519 validate_entity_state_name(entity_name, parse_entity_state(&object.body)?)?,
520 Some(object.version),
521 true,
522 )),
523 None => Ok((EntityState::new(entity_name), None, false)),
524 }
525}
526
527fn with_state_client<R, F>(
528 cloud: &mut CloudClient,
529 resolver: &StorageResolver,
530 target: &EntityStateTarget,
531 f: F,
532) -> FloeResult<R>
533where
534 F: FnOnce(&str, &dyn StorageClient) -> FloeResult<R>,
535{
536 match target {
537 EntityStateTarget::Local { uri, .. } => f(uri, &LocalClient::new()),
538 EntityStateTarget::Remote { uri, storage } => {
539 let client = cloud.client_for_context(resolver, storage, "entity state")?;
540 f(uri, client)
541 }
542 }
543}
544
545fn conditional_write_to_version(cw: ConditionalWrite) -> Option<Option<String>> {
546 match cw {
547 ConditionalWrite::Written { version } => Some(Some(version)),
548 ConditionalWrite::Conflict => None,
549 }
550}
551
552fn persist_loaded_state(
553 cloud: &mut CloudClient,
554 resolver: &StorageResolver,
555 loaded: &LoadedEntityState,
556) -> FloeResult<Option<Option<String>>> {
557 let mut state = loaded.state.clone();
558 state.schema = ENTITY_STATE_SCHEMA_V2.to_string();
559 let body = serde_json::to_vec_pretty(&state)?;
560 let version = loaded.version.as_deref();
561 let cw = with_state_client(cloud, resolver, &loaded.target, |uri, client| {
562 client.write_object_conditional(uri, version, &body)
563 })?;
564 Ok(conditional_write_to_version(cw))
565}
566
567fn delete_loaded_state(
568 cloud: &mut CloudClient,
569 resolver: &StorageResolver,
570 loaded: &LoadedEntityState,
571) -> FloeResult<Option<Option<String>>> {
572 let version = loaded.version.as_deref();
573 let cw = with_state_client(cloud, resolver, &loaded.target, |uri, client| {
574 client.delete_object_conditional(uri, version)
575 })?;
576 Ok(conditional_write_to_version(cw))
577}
578
579fn state_target_from_resolved(resolved: &ResolvedPath) -> FloeResult<EntityStateTarget> {
580 if let Some(path) = &resolved.local_path {
581 return Ok(EntityStateTarget::Local {
582 path: path.clone(),
583 uri: resolved.uri.clone(),
584 });
585 }
586 if is_remote_uri(&resolved.uri) {
587 return Ok(EntityStateTarget::Remote {
588 storage: resolved.storage.clone(),
589 uri: resolved.uri.clone(),
590 });
591 }
592 Err(Box::new(ConfigError(format!(
593 "state path is neither local nor supported remote: {}",
594 resolved.uri
595 ))))
596}
597
598fn remove_expired_claims(state: &mut EntityState) {
599 let now = time::OffsetDateTime::now_utc();
600 state.claims.retain(|_, claim| {
601 time::OffsetDateTime::parse(
602 &claim.expires_at,
603 &time::format_description::well_known::Rfc3339,
604 )
605 .map(|expires_at| expires_at > now)
606 .unwrap_or(false)
607 });
608}
609
610fn rfc3339_offset(seconds: i64) -> String {
611 (time::OffsetDateTime::now_utc() + time::Duration::seconds(seconds))
612 .format(&time::format_description::well_known::Rfc3339)
613 .unwrap_or_else(|_| crate::report::now_rfc3339())
614}
615
616fn now_rfc3339() -> String {
617 rfc3339_offset(0)
618}
619
620fn rfc3339_after_seconds(seconds: i64) -> String {
621 rfc3339_offset(seconds)
622}
623
624fn join_state_path(source_root: &str, entity_name: &str) -> String {
625 if source_root.is_empty() || source_root == "." {
626 format!(".floe/state/{entity_name}/{ENTITY_STATE_FILENAME}")
627 } else {
628 format!(
629 "{}/.floe/state/{entity_name}/{ENTITY_STATE_FILENAME}",
630 source_root.trim_end_matches(is_path_separator)
631 )
632 }
633}
634
635fn derive_source_root(
636 raw_path: &str,
637 source_format: &str,
638 resolved_local_path: Option<&Path>,
639) -> String {
640 let trimmed = raw_path.trim_end_matches(is_path_separator);
641 if trimmed.is_empty() {
642 return String::new();
643 }
644
645 if let Some(prefix) = prefix_before_first_glob(trimmed) {
646 if prefix.is_empty() {
647 return String::new();
648 }
649 if prefix.ends_with(is_path_separator) {
650 return prefix.trim_end_matches(is_path_separator).to_string();
651 }
652 return parent_like(prefix)
653 .unwrap_or(prefix)
654 .trim_end_matches(is_path_separator)
655 .to_string();
656 }
657
658 if let Some(path) = resolved_local_path.filter(|path| path.exists()) {
659 if path.is_dir() {
660 return trimmed.to_string();
661 }
662 if path.is_file() {
663 return parent_like(trimmed)
664 .unwrap_or(trimmed)
665 .trim_end_matches(is_path_separator)
666 .to_string();
667 }
668 }
669
670 if matches_source_file_suffix(trimmed, source_format) {
671 return parent_like(trimmed)
672 .unwrap_or(trimmed)
673 .trim_end_matches(is_path_separator)
674 .to_string();
675 }
676
677 trimmed.to_string()
678}
679
680fn prefix_before_first_glob(value: &str) -> Option<&str> {
681 let index = value.find(['*', '?', '['])?;
682 Some(&value[..index])
683}
684
685fn matches_source_file_suffix(value: &str, source_format: &str) -> bool {
686 let Some(segment) = value.rsplit(is_path_separator).next() else {
687 return false;
688 };
689 let segment = segment.to_ascii_lowercase();
690
691 extensions::suffixes_for_format(source_format)
692 .map(|suffixes| {
693 suffixes
694 .iter()
695 .any(|suffix| segment.ends_with(&suffix.to_ascii_lowercase()))
696 })
697 .unwrap_or(false)
698}
699
700fn parent_like(value: &str) -> Option<&str> {
701 value.rfind(is_path_separator).map(|index| {
702 if index == 0 {
703 &value[..1]
704 } else {
705 &value[..index]
706 }
707 })
708}
709
710fn is_path_separator(ch: char) -> bool {
711 ch == '/' || ch == '\\'
712}
713
714fn is_remote_uri(value: &str) -> bool {
715 value.starts_with("s3://") || value.starts_with("gs://") || value.starts_with("abfs://")
716}
717
718pub fn validate_entity_state(entity: &EntityConfig, state: EntityState) -> FloeResult<EntityState> {
719 validate_entity_state_name(&entity.name, state)
720}
721
722fn validate_entity_state_name(entity_name: &str, state: EntityState) -> FloeResult<EntityState> {
723 if state.schema != ENTITY_STATE_SCHEMA_V1 && state.schema != ENTITY_STATE_SCHEMA_V2 {
724 return Err(Box::new(ConfigError(format!(
725 "entity.name={} state schema mismatch: expected {} or {}, got {}",
726 entity_name, ENTITY_STATE_SCHEMA_V1, ENTITY_STATE_SCHEMA_V2, state.schema
727 ))));
728 }
729
730 if state.entity != entity_name {
731 return Err(Box::new(ConfigError(format!(
732 "entity.name={} state entity mismatch: expected {}, got {}",
733 entity_name, entity_name, state.entity
734 ))));
735 }
736
737 Ok(state)
738}