Skip to content

Commit 9a0c190

Browse files
Saga4Saga4
committed
fix(slack): resolve hosted polling second review
1 parent 5d02541 commit 9a0c190

5 files changed

Lines changed: 185 additions & 69 deletions

File tree

‎crates/locality-slack/fixtures/hosted-v1/provider-v1/system-subtypes-sanitized-v1.json‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
"deleted": false,
1111
"file_ids": []
1212
},
13-
"reply_count": 0
13+
"reply_count": null
1414
},
1515
{
1616
"message": {
@@ -23,7 +23,7 @@
2323
"deleted": false,
2424
"file_ids": []
2525
},
26-
"reply_count": 0
26+
"reply_count": null
2727
},
2828
{
2929
"message": {
@@ -36,6 +36,6 @@
3636
"deleted": false,
3737
"file_ids": []
3838
},
39-
"reply_count": 0
39+
"reply_count": null
4040
}
4141
]

‎crates/locality-slack/src/portable/hosted/poll.rs‎

Lines changed: 25 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -75,8 +75,6 @@ pub struct HostedSlackRepliesPageV1 {
7575
pub poll_overlap_watermark: String,
7676
pub root_message_id: String,
7777
pub root_reply_count: u32,
78-
#[serde(default, skip_serializing_if = "Option::is_none")]
79-
pub reconciliation: Option<HostedSlackRepliesReconciliationV1>,
8078
pub request_cursor: Option<String>,
8179
pub next_cursor: Option<String>,
8280
pub observed_at: String,
@@ -85,12 +83,6 @@ pub struct HostedSlackRepliesPageV1 {
8583
pub files: Vec<RawHostedSlackFileMetadata>,
8684
}
8785

88-
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
89-
#[serde(rename_all = "snake_case")]
90-
pub enum HostedSlackRepliesReconciliationV1 {
91-
ThreadNotFound,
92-
}
93-
9486
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
9587
pub enum HostedSlackPageApplyOutcomeV1 {
9688
Applied,
@@ -254,6 +246,7 @@ impl HostedSlackRepliesPageV1 {
254246
"replies page messages",
255247
));
256248
}
249+
let deleted_root_reconciliation = is_deleted_root_reconciliation_page(self);
257250
if self.request_cursor.is_none() {
258251
let first = &self.messages[0];
259252
if first.ts != self.root_message_id {
@@ -266,10 +259,11 @@ impl HostedSlackRepliesPageV1 {
266259
first.ts.clone(),
267260
));
268261
}
269-
} else if self
270-
.messages
271-
.iter()
272-
.any(|message| message.ts == self.root_message_id)
262+
} else if !deleted_root_reconciliation
263+
&& self
264+
.messages
265+
.iter()
266+
.any(|message| message.ts == self.root_message_id)
273267
{
274268
return Err(HostedSlackPollError::InvalidMessageRelationship(
275269
self.root_message_id.clone(),
@@ -288,20 +282,6 @@ impl HostedSlackRepliesPageV1 {
288282
));
289283
}
290284
}
291-
if self.reconciliation == Some(HostedSlackRepliesReconciliationV1::ThreadNotFound)
292-
&& (self.request_cursor.is_some()
293-
|| self.next_cursor.is_some()
294-
|| self.root_reply_count != 0
295-
|| self.messages.len() != 1
296-
|| self.messages[0].ts != self.root_message_id
297-
|| !self.messages[0].deleted
298-
|| !self.messages[0].text.is_empty()
299-
|| !self.messages[0].file_ids.is_empty())
300-
{
301-
return Err(HostedSlackPollError::IncompleteCandidate(
302-
"thread-not-found reconciliation",
303-
));
304-
}
305285
validate_serialized_page_size(self, "replies page")?;
306286
Ok(())
307287
}
@@ -473,8 +453,7 @@ impl HostedSlackPollCheckpointV1 {
473453
return Err(HostedSlackPollError::PageWindowMismatch);
474454
}
475455
let first_page = page.request_cursor.is_none();
476-
let deleted_root_reconciliation =
477-
page.reconciliation == Some(HostedSlackRepliesReconciliationV1::ThreadNotFound);
456+
let deleted_root_reconciliation = is_deleted_root_reconciliation_page(page);
478457
let expectation_is_from_this_catch_up = page.phase
479458
!= HostedSlackPollPhaseV1::CatchUpReplies
480459
|| catch_up_history_contains_root(&next, &page.root_message_id)?
@@ -505,7 +484,7 @@ impl HostedSlackPollCheckpointV1 {
505484
},
506485
);
507486
}
508-
if first_page {
487+
if first_page || deleted_root_reconciliation {
509488
next.candidate.messages.retain(|message| {
510489
normalized_root_id(message) != Some(page.root_message_id.as_str())
511490
});
@@ -1645,6 +1624,23 @@ fn normalized_root_id(message: &RawHostedSlackMessage) -> Option<&str> {
16451624
.filter(|thread_ts| *thread_ts != message.ts)
16461625
}
16471626

1627+
// V1 reconciliation deliberately uses only the existing replies-page schema so
1628+
// deny-unknown V1 readers never receive an unversioned field. The exact terminal
1629+
// tombstone shape is reserved for a root Slack reports as thread_not_found.
1630+
fn is_deleted_root_reconciliation_page(page: &HostedSlackRepliesPageV1) -> bool {
1631+
page.next_cursor.is_none()
1632+
&& page.root_reply_count == 0
1633+
&& page.messages.len() == 1
1634+
&& page.messages[0].ts == page.root_message_id
1635+
&& normalized_root_id(&page.messages[0]).is_none()
1636+
&& page.messages[0].deleted
1637+
&& page.messages[0].text.is_empty()
1638+
&& page.messages[0].edited_ts.is_none()
1639+
&& page.messages[0].file_ids.is_empty()
1640+
&& page.users.is_empty()
1641+
&& page.files.is_empty()
1642+
}
1643+
16481644
fn record_page(
16491645
checkpoint: &mut HostedSlackPollCheckpointV1,
16501646
phase: HostedSlackPollPhaseV1,

‎crates/locality-slack/src/portable/hosted/provider.rs‎

Lines changed: 84 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,8 @@ use super::native::{
3030
use super::poll::{
3131
HOSTED_SLACK_POLL_PAGE_FORMAT_VERSION_V1, HOSTED_SLACK_POLL_PAGE_MINIMUM_READER_VERSION_V1,
3232
HostedSlackHistoryMessageV1, HostedSlackHistoryPageV1, HostedSlackPollOutputV1,
33-
HostedSlackRepliesPageV1, HostedSlackRepliesReconciliationV1,
34-
hosted_slack_history_page_reference_closure_v1, hosted_slack_replies_page_reference_closure_v1,
33+
HostedSlackRepliesPageV1, hosted_slack_history_page_reference_closure_v1,
34+
hosted_slack_replies_page_reference_closure_v1,
3535
};
3636

3737
pub const HOSTED_SLACK_PROVIDER_PAGE_LIMIT_V1: u32 = 15;
@@ -666,7 +666,7 @@ pub async fn drive_hosted_slack_poll_v1<P: HostedSlackProviderPort>(
666666
.await
667667
{
668668
Ok(response) => response,
669-
Err(HostedSlackProviderError::ThreadNotFound) if request.cursor.is_none() => {
669+
Err(HostedSlackProviderError::ThreadNotFound) => {
670670
let page = deleted_root_reconciliation_page(checkpoint, &request)?;
671671
ensure_active(control)?;
672672
checkpoint.apply_replies_page(&page)?;
@@ -887,11 +887,7 @@ fn history_poll_page(
887887
.map(|provided| {
888888
let is_root = normalized_root_id(&provided.message).is_none();
889889
let reply_count = if is_root {
890-
provided
891-
.reply_count
892-
.ok_or(HostedSlackProviderError::InvalidResponse(
893-
"root reply_count",
894-
))?
890+
provided.reply_count.unwrap_or(0)
895891
} else {
896892
if provided.reply_count.is_some_and(|count| count != 0) {
897893
return Err(HostedSlackProviderError::InvalidResponse(
@@ -979,7 +975,6 @@ fn replies_poll_page(
979975
poll_overlap_watermark: checkpoint.poll_overlap_watermark().to_string(),
980976
root_message_id: request.root_message_id.clone(),
981977
root_reply_count,
982-
reconciliation: None,
983978
request_cursor: request.cursor.clone(),
984979
next_cursor,
985980
observed_at: response.observed_at,
@@ -1028,8 +1023,7 @@ fn deleted_root_reconciliation_page(
10281023
poll_overlap_watermark: checkpoint.poll_overlap_watermark().to_string(),
10291024
root_message_id: request.root_message_id.clone(),
10301025
root_reply_count: 0,
1031-
reconciliation: Some(HostedSlackRepliesReconciliationV1::ThreadNotFound),
1032-
request_cursor: None,
1026+
request_cursor: request.cursor.clone(),
10331027
next_cursor: None,
10341028
observed_at: current_canonical_utc(),
10351029
messages: vec![root],
@@ -1231,11 +1225,17 @@ where
12311225
if attempt == retry.max_retries {
12321226
return Err(error);
12331227
}
1234-
if delay > MAX_HOSTED_SLACK_PROVIDER_RETRY_AFTER_V1 {
1235-
wait_for_retry(control, MAX_HOSTED_SLACK_PROVIDER_RETRY_AFTER_V1).await?;
1228+
let in_drive_wait = delay.min(MAX_HOSTED_SLACK_PROVIDER_RETRY_AFTER_V1);
1229+
if control
1230+
.remaining()
1231+
.is_none_or(|remaining| in_drive_wait >= remaining)
1232+
{
1233+
return Err(error);
1234+
}
1235+
wait_for_retry(control, in_drive_wait).await?;
1236+
if delay > in_drive_wait {
12361237
return Err(error);
12371238
}
1238-
wait_for_retry(control, delay).await?;
12391239
}
12401240
}
12411241
}
@@ -1368,15 +1368,43 @@ struct HostedSlackMethodGateState {
13681368
in_flight: usize,
13691369
tokens: f64,
13701370
last_refill: Instant,
1371-
cooldown_until: Option<Instant>,
1371+
cooldown: Option<HostedSlackMethodCooldown>,
1372+
}
1373+
1374+
struct HostedSlackMethodCooldown {
1375+
started_at: Instant,
1376+
duration: Duration,
1377+
checked_until: Option<Instant>,
1378+
}
1379+
1380+
impl HostedSlackMethodCooldown {
1381+
fn new(started_at: Instant, duration: Duration) -> Self {
1382+
Self {
1383+
started_at,
1384+
duration,
1385+
checked_until: started_at.checked_add(duration),
1386+
}
1387+
}
1388+
1389+
fn remaining(&self, now: Instant) -> Duration {
1390+
self.checked_until.map_or_else(
1391+
|| {
1392+
self.duration
1393+
.saturating_sub(now.saturating_duration_since(self.started_at))
1394+
},
1395+
|until| until.saturating_duration_since(now),
1396+
)
1397+
}
13721398
}
13731399

13741400
impl HostedSlackMethodGateState {
13751401
fn refill(&mut self, config: &ConnectorNetworkConfig, now: Instant) {
1376-
if self.cooldown_until.is_some_and(|until| until > now) {
1402+
if let Some(cooldown) = &self.cooldown
1403+
&& !cooldown.remaining(now).is_zero()
1404+
{
13771405
return;
13781406
}
1379-
self.cooldown_until = None;
1407+
self.cooldown = None;
13801408
let elapsed = now.saturating_duration_since(self.last_refill);
13811409
self.tokens =
13821410
(self.tokens + elapsed.as_secs_f64() * config.requests_per_second).min(config.burst);
@@ -1401,7 +1429,7 @@ impl HostedSlackMethodGate {
14011429
in_flight: 0,
14021430
tokens,
14031431
last_refill: Instant::now(),
1404-
cooldown_until: None,
1432+
cooldown: None,
14051433
}),
14061434
changed,
14071435
}),
@@ -1425,8 +1453,12 @@ impl HostedSlackMethodGate {
14251453
inner: self.inner.clone(),
14261454
};
14271455
}
1428-
if let Some(until) = state.cooldown_until.filter(|until| *until > now) {
1429-
Some(until.saturating_duration_since(now))
1456+
if let Some(cooldown) = &state.cooldown {
1457+
Some(
1458+
cooldown
1459+
.remaining(now)
1460+
.min(MAX_HOSTED_SLACK_PROVIDER_RETRY_AFTER_V1),
1461+
)
14301462
} else if state.in_flight < self.inner.config.max_in_flight {
14311463
let missing = (1.0 - state.tokens).max(0.0);
14321464
Some(Duration::from_secs_f64(
@@ -1455,14 +1487,16 @@ impl HostedSlackMethodGate {
14551487

14561488
fn record_cooldown(&self, delay: Duration) {
14571489
let now = Instant::now();
1458-
let until = now + delay;
1490+
let candidate = HostedSlackMethodCooldown::new(now, delay);
14591491
{
14601492
let mut state = self.inner.state.lock().expect("hosted Slack gate lock");
1461-
state.cooldown_until = Some(
1462-
state
1463-
.cooldown_until
1464-
.map_or(until, |current| current.max(until)),
1465-
);
1493+
if state
1494+
.cooldown
1495+
.as_ref()
1496+
.is_none_or(|current| candidate.remaining(now) > current.remaining(now))
1497+
{
1498+
state.cooldown = Some(candidate);
1499+
}
14661500
state.tokens = 0.0;
14671501
state.last_refill = now;
14681502
}
@@ -1479,9 +1513,10 @@ impl HostedSlackMethodGate {
14791513
in_flight: state.in_flight,
14801514
tokens: state.tokens,
14811515
cooldown_remaining: state
1482-
.cooldown_until
1483-
.filter(|until| *until > now)
1484-
.map(|until| until.saturating_duration_since(now)),
1516+
.cooldown
1517+
.as_ref()
1518+
.map(|cooldown| cooldown.remaining(now))
1519+
.filter(|remaining| !remaining.is_zero()),
14851520
}
14861521
}
14871522
}
@@ -2441,7 +2476,6 @@ fn provider_message(
24412476
body.text = bounded_system_event_text(subtype, &body.text);
24422477
body.edited = None;
24432478
body.files.clear();
2444-
body.reply_count = Some(0);
24452479
(body, false)
24462480
}
24472481
Some(_) => {
@@ -3138,8 +3172,19 @@ mod tests {
31383172
assert!(bounded.message.text.ends_with(" …[truncated]"));
31393173
assert!(std::str::from_utf8(bounded.message.text.as_bytes()).is_ok());
31403174

3175+
let threaded = serde_json::from_str::<MessageWire>(
3176+
r#"{"ts":"1780000004.000100","subtype":"channel_topic","text":"topic","reply_count":2}"#,
3177+
)
3178+
.unwrap();
3179+
assert_eq!(
3180+
provider_message(threaded, "C08ENGINEER1")
3181+
.unwrap()
3182+
.reply_count,
3183+
Some(2)
3184+
);
3185+
31413186
let unknown = serde_json::from_str::<MessageWire>(
3142-
r#"{"ts":"1780000004.000100","subtype":"unknown_provider_secret","text":"secret"}"#,
3187+
r#"{"ts":"1780000005.000100","subtype":"unknown_provider_secret","text":"secret"}"#,
31433188
)
31443189
.unwrap();
31453190
assert_eq!(
@@ -3811,6 +3856,14 @@ mod tests {
38113856
MAX_HOSTED_SLACK_PROVIDER_RETRY_AFTER_V1,
38123857
Duration::from_secs(301),
38133858
),
3859+
(
3860+
vec![("Retry-After", "18446744073709551615")],
3861+
HostedSlackProviderError::RateLimited {
3862+
retry_after: Duration::from_secs(u64::MAX),
3863+
},
3864+
Duration::from_secs(u64::MAX - 1),
3865+
Duration::from_secs(u64::MAX),
3866+
),
38143867
];
38153868

38163869
for (headers, expected, minimum_cooldown, maximum_cooldown) in cases {

‎crates/locality-slack/tests/hosted_polling.rs‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1036,6 +1036,16 @@ fn page_and_checkpoint_decoders_enforce_bounds_unknown_fields_and_scope() {
10361036
Err(HostedSlackPollError::InvalidJson("history page"))
10371037
);
10381038

1039+
let mut unversioned_reconciliation =
1040+
serde_json::from_slice::<serde_json::Value>(REPLIES_PAGE).unwrap();
1041+
unversioned_reconciliation["reconciliation"] = serde_json::json!("thread_not_found");
1042+
assert_eq!(
1043+
decode_hosted_slack_replies_page_v1(
1044+
&serde_json::to_vec(&unversioned_reconciliation).unwrap()
1045+
),
1046+
Err(HostedSlackPollError::InvalidJson("replies page"))
1047+
);
1048+
10391049
let mut too_many = history_page();
10401050
too_many.messages =
10411051
vec![too_many.messages[0].clone(); MAX_HOSTED_SLACK_POLL_PAGE_MESSAGES_V1 + 1];

0 commit comments

Comments
 (0)