Skip to content

Commit 5e3f7a4

Browse files
tomasoltomasol
committed
r: Extract PendingStatePendingAt and PendingStateBlockedByJoinSet 
1 parent 539db52 commit 5e3f7a4

11 files changed

Lines changed: 130 additions & 105 deletions

File tree

‎crates/concepts/src/storage.rs‎

Lines changed: 24 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1647,22 +1647,12 @@ pub enum PendingState {
16471647
/// Caused by [`ExecutionRequest::Locked`].
16481648
Locked(PendingStateLocked),
16491649

1650-
#[display("PendingAt(`{scheduled_at}`)")]
1651-
PendingAt {
1652-
scheduled_at: DateTime<Utc>,
1653-
/// `last_lock` is needed for lock extension.
1654-
last_lock: Option<LockedBy>,
1655-
},
1650+
#[display("PendingAt(`{_0}`)")]
1651+
PendingAt(PendingStatePendingAt),
16561652

16571653
/// Caused by [`HistoryEvent::JoinNext`]
1658-
#[display("BlockedByJoinSet({join_set_id},`{lock_expires_at}`)")]
1659-
BlockedByJoinSet {
1660-
join_set_id: JoinSetId,
1661-
/// See [`HistoryEvent::JoinNext::lock_expires_at`].
1662-
lock_expires_at: DateTime<Utc>,
1663-
/// Blocked by closing of the join set
1664-
closing: bool,
1665-
},
1654+
#[display("BlockedByJoinSet({_0})")]
1655+
BlockedByJoinSet(PendingStateBlockedByJoinSet),
16661656

16671657
#[display("Paused")]
16681658
Paused,
@@ -1681,6 +1671,24 @@ pub struct PendingStateLocked {
16811671
pub lock_expires_at: DateTime<Utc>,
16821672
}
16831673

1674+
#[derive(Debug, Clone, derive_more::Display, PartialEq, Eq, Serialize, Deserialize)]
1675+
#[display("`{scheduled_at}`, last_lock={last_lock:?}")]
1676+
pub struct PendingStatePendingAt {
1677+
pub scheduled_at: DateTime<Utc>,
1678+
/// `last_lock` is needed for lock extension.
1679+
pub last_lock: Option<LockedBy>,
1680+
}
1681+
1682+
#[derive(Debug, Clone, derive_more::Display, PartialEq, Eq, Serialize, Deserialize)]
1683+
#[display("{join_set_id}, `{lock_expires_at}`, closing={closing}")]
1684+
pub struct PendingStateBlockedByJoinSet {
1685+
pub join_set_id: JoinSetId,
1686+
/// See [`HistoryEvent::JoinNext::lock_expires_at`].
1687+
pub lock_expires_at: DateTime<Utc>,
1688+
/// Blocked by closing of the join set
1689+
pub closing: bool,
1690+
}
1691+
16841692
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
16851693
pub struct LockedBy {
16861694
pub executor_id: ExecutorId,
@@ -1756,10 +1764,10 @@ impl PendingState {
17561764
));
17571765
}
17581766
match self {
1759-
PendingState::PendingAt {
1767+
PendingState::PendingAt(PendingStatePendingAt {
17601768
scheduled_at,
17611769
last_lock,
1762-
} => {
1770+
}) => {
17631771
if *scheduled_at <= created_at {
17641772
// pending now, ok to lock
17651773
Ok(LockKind::CreatingNewLock)

‎crates/db-mem/src/inmemory_dao.rs‎

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -460,7 +460,9 @@ mod index {
460460
use crate::journal::ExecutionJournal;
461461
use concepts::component_id::InputContentDigest;
462462
use concepts::prefixed_ulid::DelayId;
463-
use concepts::storage::{HistoryEvent, JoinSetRequest, JoinSetResponse, PendingStateLocked};
463+
use concepts::storage::{
464+
HistoryEvent, JoinSetRequest, JoinSetResponse, PendingStateLocked, PendingStatePendingAt,
465+
};
464466
use tracing::trace;
465467

466468
#[derive(Debug, Default)]
@@ -547,10 +549,10 @@ mod index {
547549
self.purge(execution_id);
548550
// Add it again if needed
549551
match journal.pending_state {
550-
PendingState::PendingAt {
552+
PendingState::PendingAt(PendingStatePendingAt {
551553
scheduled_at,
552554
last_lock: _,
553-
} => {
555+
}) => {
554556
self.pending_scheduled
555557
.entry(scheduled_at)
556558
.or_default()
@@ -570,7 +572,7 @@ mod index {
570572
.or_default()
571573
.push(lock_expires_at);
572574
}
573-
PendingState::BlockedByJoinSet { .. }
575+
PendingState::BlockedByJoinSet(..)
574576
| PendingState::Finished { .. }
575577
| PendingState::Paused => {}
576578
}
@@ -886,7 +888,7 @@ impl DbHolder {
886888
}
887889
let next_version = journal.append(created_at, event, appending_version)?;
888890
self.index.update(journal);
889-
if matches!(journal.pending_state, PendingState::PendingAt { .. })
891+
if matches!(journal.pending_state, PendingState::PendingAt(..))
890892
&& let Some(subscription) = self.ffqn_to_pending_subscription.get(journal.ffqn())
891893
{
892894
let _ = subscription.try_send(());
@@ -997,7 +999,7 @@ impl DbHolder {
997999
}
9981000
let version = journal.version();
9991001
self.index.update(journal);
1000-
if matches!(journal.pending_state, PendingState::PendingAt { .. })
1002+
if matches!(journal.pending_state, PendingState::PendingAt(..))
10011003
&& let Some(subscription) = self.ffqn_to_pending_subscription.get(journal.ffqn())
10021004
{
10031005
let _ = subscription.try_send(());
@@ -1053,7 +1055,7 @@ impl DbHolder {
10531055
};
10541056
journal.append_response(response_event.created_at, response_event.event)?;
10551057
self.index.update(journal);
1056-
if matches!(journal.pending_state, PendingState::PendingAt { .. })
1058+
if matches!(journal.pending_state, PendingState::PendingAt(..))
10571059
&& let Some(subscription) = self.ffqn_to_pending_subscription.get(journal.ffqn())
10581060
{
10591061
let _ = subscription.try_send(());

‎crates/db-mem/src/journal.rs‎

Lines changed: 22 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,9 @@ use concepts::prefixed_ulid::DeploymentId;
33
use concepts::storage::{
44
CreateRequest, DbErrorWrite, DbErrorWriteNonRetriable, ExecutionEvent, ExecutionRequest,
55
HistoryEvent, JoinSetRequest, JoinSetResponse, JoinSetResponseEvent, JoinSetResponseEventOuter,
6-
Locked, LockedBy, PendingStateFinished, PendingStateFinishedResultKind, PendingStateLocked,
7-
ResponseCursor, ResponseWithCursor, VersionType,
6+
Locked, LockedBy, PendingStateBlockedByJoinSet, PendingStateFinished,
7+
PendingStateFinishedResultKind, PendingStateLocked, PendingStatePendingAt, ResponseCursor,
8+
ResponseWithCursor, VersionType,
89
};
910
use concepts::storage::{ExecutionLog, PendingState, Version};
1011
use concepts::{ComponentId, JoinSetId};
@@ -29,10 +30,10 @@ pub(crate) struct ExecutionJournal {
2930
impl ExecutionJournal {
3031
#[must_use]
3132
pub fn new(req: CreateRequest) -> Self {
32-
let pending_state = PendingState::PendingAt {
33+
let pending_state = PendingState::PendingAt(PendingStatePendingAt {
3334
scheduled_at: req.scheduled_at,
3435
last_lock: None,
35-
};
36+
});
3637
let execution_id = req.execution_id.clone();
3738
let component_id = req.component_id.clone();
3839
let deployment_id = req.deployment_id;
@@ -305,10 +306,12 @@ impl ExecutionJournal {
305306
.enumerate()
306307
.rev()
307308
.find_map(|(idx, event)| match &event.event {
308-
ExecutionRequest::Created { scheduled_at, .. } => Some(PendingState::PendingAt {
309-
scheduled_at: *scheduled_at,
310-
last_lock: None,
311-
}),
309+
ExecutionRequest::Created { scheduled_at, .. } => {
310+
Some(PendingState::PendingAt(PendingStatePendingAt {
311+
scheduled_at: *scheduled_at,
312+
last_lock: None,
313+
}))
314+
}
312315

313316
ExecutionRequest::Finished { result, .. } => {
314317
assert_eq!(self.execution_events.len() - 1, idx);
@@ -347,10 +350,10 @@ impl ExecutionJournal {
347350
| ExecutionRequest::Unlocked {
348351
backoff_expires_at: expires_at,
349352
..
350-
} => Some(PendingState::PendingAt {
353+
} => Some(PendingState::PendingAt(PendingStatePendingAt {
351354
scheduled_at: *expires_at,
352355
last_lock: self.find_last_lock().map(LockedBy::from),
353-
}),
356+
})),
354357

355358
ExecutionRequest::HistoryEvent {
356359
event:
@@ -390,17 +393,19 @@ impl ExecutionJournal {
390393
if let Some(nth_created_at) = resp {
391394
// Original executor has a chance to continue, but after expiry any executor can pick up the execution.
392395
let scheduled_at = max(*lock_expires_at, *nth_created_at);
393-
Some(PendingState::PendingAt {
396+
Some(PendingState::PendingAt(PendingStatePendingAt {
394397
scheduled_at,
395398
last_lock: self.find_last_lock().map(LockedBy::from),
396-
})
399+
}))
397400
} else {
398401
// Still waiting for response
399-
Some(PendingState::BlockedByJoinSet {
400-
join_set_id: expected_join_set_id.clone(),
401-
lock_expires_at: *lock_expires_at,
402-
closing: *closing,
403-
})
402+
Some(PendingState::BlockedByJoinSet(
403+
PendingStateBlockedByJoinSet {
404+
join_set_id: expected_join_set_id.clone(),
405+
lock_expires_at: *lock_expires_at,
406+
closing: *closing,
407+
},
408+
))
404409
}
405410
}
406411
ExecutionRequest::Unpaused => {

‎crates/db-postgres/src/postgres_dao.rs‎

Lines changed: 12 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,10 @@ use concepts::{
1717
JoinSetResponse, JoinSetResponseEvent, JoinSetResponseEventOuter, ListExecutionsFilter,
1818
ListLogsResponse, LockPendingResponse, Locked, LockedBy, LockedExecution, LogEntry,
1919
LogEntryRow, LogFilter, LogInfoAppendRow, LogLevel, LogStreamType, Pagination,
20-
PendingState, PendingStateFinished, PendingStateFinishedResultKind, PendingStateLocked,
21-
ResponseCursor, ResponseWithCursor, STATE_BLOCKED_BY_JOIN_SET, STATE_FINISHED,
22-
STATE_LOCKED, STATE_PENDING_AT, TimeoutOutcome, Version, VersionType, WasmBacktrace,
20+
PendingState, PendingStateBlockedByJoinSet, PendingStateFinished,
21+
PendingStateFinishedResultKind, PendingStateLocked, PendingStatePendingAt, ResponseCursor,
22+
ResponseWithCursor, STATE_BLOCKED_BY_JOIN_SET, STATE_FINISHED, STATE_LOCKED,
23+
STATE_PENDING_AT, TimeoutOutcome, Version, VersionType, WasmBacktrace,
2324
},
2425
};
2526
use deadpool_postgres::{Client, ManagerConfig, Pool, RecyclingMethod};
@@ -675,10 +676,10 @@ impl CombinedState {
675676
ffqn,
676677
created_at,
677678
first_scheduled_at,
678-
pending_state: PendingState::PendingAt {
679+
pending_state: PendingState::PendingAt(PendingStatePendingAt {
679680
scheduled_at,
680681
last_lock: None,
681-
},
682+
}),
682683
},
683684
// Pending, previously locked
684685
CombinedStateDTO {
@@ -706,13 +707,13 @@ impl CombinedState {
706707
ffqn,
707708
created_at,
708709
first_scheduled_at,
709-
pending_state: PendingState::PendingAt {
710+
pending_state: PendingState::PendingAt(PendingStatePendingAt {
710711
scheduled_at,
711712
last_lock: Some(LockedBy {
712713
executor_id,
713714
run_id,
714715
}),
715-
},
716+
}),
716717
},
717718
CombinedStateDTO {
718719
execution_id,
@@ -772,11 +773,11 @@ impl CombinedState {
772773
ffqn,
773774
created_at,
774775
first_scheduled_at,
775-
pending_state: PendingState::BlockedByJoinSet {
776+
pending_state: PendingState::BlockedByJoinSet(PendingStateBlockedByJoinSet {
776777
join_set_id: join_set_id.clone(),
777778
closing: join_set_closing,
778779
lock_expires_at,
779-
},
780+
}),
780781
},
781782
CombinedStateDTO {
782783
execution_id,
@@ -2715,11 +2716,11 @@ async fn append_response(
27152716
let combined_state = get_combined_state(tx, execution_id).await?;
27162717
debug!("previous_pending_state: {combined_state:?}");
27172718

2718-
let mut notifier = if let PendingState::BlockedByJoinSet {
2719+
let mut notifier = if let PendingState::BlockedByJoinSet(PendingStateBlockedByJoinSet {
27192720
join_set_id: found_join_set_id,
27202721
lock_expires_at,
27212722
closing: _,
2722-
} = combined_state.execution_with_state.pending_state
2723+
}) = combined_state.execution_with_state.pending_state
27232724
&& *join_set_id == found_join_set_id
27242725
{
27252726
let scheduled_at = std::cmp::max(lock_expires_at, event.created_at);

‎crates/db-sqlite/src/sqlite_dao.rs‎

Lines changed: 12 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,10 @@ use concepts::{
1717
JoinSetResponse, JoinSetResponseEvent, JoinSetResponseEventOuter, ListExecutionsFilter,
1818
ListLogsResponse, LockPendingResponse, Locked, LockedBy, LockedExecution, LogEntry,
1919
LogEntryRow, LogFilter, LogInfoAppendRow, LogLevel, LogStreamType, Pagination,
20-
PendingState, PendingStateFinished, PendingStateFinishedResultKind, PendingStateLocked,
21-
ResponseCursor, ResponseWithCursor, STATE_BLOCKED_BY_JOIN_SET, STATE_FINISHED,
22-
STATE_LOCKED, STATE_PENDING_AT, TimeoutOutcome, Version, VersionType,
20+
PendingState, PendingStateBlockedByJoinSet, PendingStateFinished,
21+
PendingStateFinishedResultKind, PendingStateLocked, PendingStatePendingAt, ResponseCursor,
22+
ResponseWithCursor, STATE_BLOCKED_BY_JOIN_SET, STATE_FINISHED, STATE_LOCKED,
23+
STATE_PENDING_AT, TimeoutOutcome, Version, VersionType,
2324
},
2425
};
2526
use const_format::formatcp;
@@ -585,10 +586,10 @@ impl CombinedState {
585586
ffqn,
586587
created_at,
587588
first_scheduled_at,
588-
pending_state: PendingState::PendingAt {
589+
pending_state: PendingState::PendingAt(PendingStatePendingAt {
589590
scheduled_at,
590591
last_lock: None,
591-
},
592+
}),
592593
},
593594
// Pending, previously locked
594595
CombinedStateDTO {
@@ -616,13 +617,13 @@ impl CombinedState {
616617
ffqn,
617618
created_at,
618619
first_scheduled_at,
619-
pending_state: PendingState::PendingAt {
620+
pending_state: PendingState::PendingAt(PendingStatePendingAt {
620621
scheduled_at,
621622
last_lock: Some(LockedBy {
622623
executor_id,
623624
run_id,
624625
}),
625-
},
626+
}),
626627
},
627628
CombinedStateDTO {
628629
execution_id,
@@ -682,11 +683,11 @@ impl CombinedState {
682683
ffqn,
683684
created_at,
684685
first_scheduled_at,
685-
pending_state: PendingState::BlockedByJoinSet {
686+
pending_state: PendingState::BlockedByJoinSet(PendingStateBlockedByJoinSet {
686687
join_set_id: join_set_id.clone(),
687688
closing: join_set_closing,
688689
lock_expires_at,
689-
},
690+
}),
690691
},
691692
CombinedStateDTO {
692693
execution_id,
@@ -2814,11 +2815,11 @@ impl SqlitePool {
28142815
// if the execution is going to be unblocked by this response...
28152816
let combined_state = Self::get_combined_state(tx, execution_id)?;
28162817
debug!("previous_pending_state: {combined_state:?}");
2817-
let mut notifier = if let PendingState::BlockedByJoinSet {
2818+
let mut notifier = if let PendingState::BlockedByJoinSet(PendingStateBlockedByJoinSet {
28182819
join_set_id: found_join_set_id,
28192820
lock_expires_at, // Set to a future time if the worker is keeping the execution warm waiting for the result.
28202821
closing: _,
2821-
} = combined_state.execution_with_state.pending_state
2822+
}) = combined_state.execution_with_state.pending_state
28222823
&& *join_set_id == found_join_set_id
28232824
{
28242825
// PendingAt should be set to current time if called from expired_timers_watcher,

‎crates/executor/src/executor.rs‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -880,7 +880,9 @@ mod tests {
880880
CreateRequest, DbConnectionTest, JoinSetRequest, JoinSetResponse, JoinSetResponseEvent,
881881
};
882882
use concepts::storage::{DbPoolCloseable, LockedBy};
883-
use concepts::storage::{ExecutionEvent, ExecutionRequest, HistoryEvent, PendingState};
883+
use concepts::storage::{
884+
ExecutionEvent, ExecutionRequest, HistoryEvent, PendingState, PendingStatePendingAt,
885+
};
884886
use concepts::time::{ConstClock, Now};
885887
use concepts::{
886888
FunctionMetadata, JoinSetKind, ParameterTypes, Params, RETURN_TYPE_DUMMY,
@@ -1549,10 +1551,10 @@ mod tests {
15491551
.unwrap();
15501552
assert_matches!(
15511553
parent_log.pending_state,
1552-
PendingState::PendingAt {
1554+
PendingState::PendingAt(PendingStatePendingAt {
15531555
scheduled_at,
15541556
last_lock: Some(LockedBy { executor_id: found_executor_id, run_id: _}),
1555-
} if scheduled_at == sim_clock.now() && found_executor_id == parent_executor_id,
1557+
}) if scheduled_at == sim_clock.now() && found_executor_id == parent_executor_id,
15561558
"parent should be back to pending"
15571559
);
15581560
let (found_join_set_id, found_child_execution_id, child_finished_version, found_result) = assert_matches!(
@@ -1729,13 +1731,13 @@ mod tests {
17291731
);
17301732
assert_matches!(
17311733
execution_log.pending_state,
1732-
PendingState::PendingAt {
1734+
PendingState::PendingAt(PendingStatePendingAt {
17331735
scheduled_at: found_scheduled_by,
17341736
last_lock: Some(LockedBy {
17351737
executor_id: found_executor_id,
17361738
run_id: _,
17371739
}),
1738-
} if found_scheduled_by == expected_first_timeout_expiry && found_executor_id == exec_config.executor_id
1740+
}) if found_scheduled_by == expected_first_timeout_expiry && found_executor_id == exec_config.executor_id
17391741
);
17401742
sim_clock.move_time_forward(timeout_duration);
17411743
let now_after_first_timeout = sim_clock.now();

0 commit comments

Comments
 (0)