From 412ce7cbac6fa665c3b30cde8852854295b4aba8 Mon Sep 17 00:00:00 2001 From: jgjesdal Date: Fri, 18 Sep 2026 10:02:42 +0200 Subject: [PATCH] fix(binary): read the stale-series retry from the typed problem is_stale_series_rejection searched the raw body for "unknown-timeseries" or "external-id-mismatch", the idiom from before ResponseError::problem() existed. That also matched the SDK's own pre-flight 404, which names the series it could not resolve, so an external id spelling one of those tokens earned a pointless second attempt. It reads the problem type now. Platform #120 gave each binary refusal the type its status calls for, so the slug discriminates here the way it does on every other endpoint: unknown-timeseries and external-id-mismatch are the two that a re-resolve can fix. The single datapoint-block-rejected that preceded them is deliberately not matched, and a test says so: the binary path has never been in a release, so nothing can be sending it. The retry had no live coverage because the SDK resolves every series before sending. test_insert_datapoints_binary_re_resolves_a_recreated_series deletes and recreates a series under the same external id, so the server refuses the cached id. It needs a backend at platform #120 or later, hence the #[ignore], next to test_datapoints_binary which needs the endpoint at all. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: jgjesdal --- AGENTS.md | 2 +- README.md | 8 +++++- src/timeseries/binary.rs | 56 +++++++++++++++++++++++++++++++++++++--- src/timeseries/test.rs | 51 ++++++++++++++++++++++++++++++++++++ 4 files changed, 111 insertions(+), 6 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index b984fa8..66cbe6d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -143,7 +143,7 @@ When a datapoint/event send can't get through, ingestion spools to a segmented, ### Binary datapoint ingest (`src/timeseries/binary.rs`) -`TimeSeriesService::insert_datapoints_binary` is the second ingest path, to `POST /timeseries/data/binary`: the same `DatapointsCollection` input, resolved through `/timeseries/byids` (cached per service instance; needs read access on the dataset), checked against each series' value type locally, cut into Arrow IPC frames at the contract's caps, zstd-compressed per frame (mandatory: level 1, 3 or 9, default 9) and posted through `execute_post_bytes_request`. `FrameWriter` builds one frame and is public; `cut_into_writers` and `pack_requests` hold the caps. The byte layout is the platform's `binary_datapoints_format.md`. The arrow-rs crates (`arrow-array`, `arrow-schema`, `arrow-ipc`) exist for this path and are the seed of the Arrow read path. Not in the Python bindings yet. The ignored `timeseries::tests::test_datapoints_binary` is the live twin of `test_datapoints` and needs a backend that serves the endpoint; the writer's own tests in `binary.rs` run offline. +`TimeSeriesService::insert_datapoints_binary` is the second ingest path, to `POST /timeseries/data/binary`: the same `DatapointsCollection` input, resolved through `/timeseries/byids` (cached per service instance; needs read access on the dataset), checked against each series' value type locally, cut into Arrow IPC frames at the contract's caps, zstd-compressed per frame (mandatory: level 1, 3 or 9, default 9) and posted through `execute_post_bytes_request`. Every refusal on this path is a problem document typed for its status — `invalid-frame` (400), `unknown-timeseries` (404), `request-too-large` (413), `unsupported-media-type` (415), `value-type-mismatch` and `external-id-mismatch` (422), `too-many-in-flight` (429) — with the kebab-case sub-case in a `reason` extension beside it. It was one type, `datapoint-block-rejected`, answering with all six statuses until the api split it (platform #120); the SDK matches the new slugs only, and nothing here recognises the old one — the binary path has never been in a release, so no caller can be on the other side of that change. `is_stale_series_rejection` reads the slug to decide the one re-resolve-and-retry, and reads it from the problem rather than the body text because the SDK's own pre-flight 404 names the series it could not resolve — an external id spelling `unknown-timeseries` used to trigger the retry. `FrameWriter` builds one frame and is public; `cut_into_writers` and `pack_requests` hold the caps. The byte layout is the platform's `binary_datapoints_format.md`. The arrow-rs crates (`arrow-array`, `arrow-schema`, `arrow-ipc`) exist for this path and are the seed of the Arrow read path. Not in the Python bindings yet. The ignored `timeseries::tests::test_datapoints_binary` is the live twin of `test_datapoints` and needs a backend that serves the endpoint; the writer's own tests in `binary.rs` run offline. ### The `ApiServiceProvider` trait (`src/generic.rs`) diff --git a/README.md b/README.md index 9ae94e0..0653c0b 100644 --- a/README.md +++ b/README.md @@ -128,7 +128,13 @@ frame, 32 frames per request), compressed with zstd (level 9 by default, 1 and 3 choices) and posted. A 204 means every frame was accepted. A series that does not exist is a 404 before anything is sent, a value that does not fit its -type is a 422, and a 429 or a 5xx is retried. Resolving by external id goes through +type is a 422, and a 429 or a 5xx is retried. Every refusal the server sends is a problem +document with its own type — `invalid-frame`, `unknown-timeseries`, `value-type-mismatch`, +`external-id-mismatch`, `too-many-in-flight`, `request-too-large`, `unsupported-media-type` — +so branch on `problem_slug()`; the kebab-case `reason` extension beside it names the exact +sub-case, which is the only way to tell thirteen kinds of malformed frame apart. A series +deleted or renamed since it was cached is the one refusal the SDK handles itself: it re-resolves +and sends again, once. Resolving by external id goes through `/timeseries/byids`, so the caller needs read access on the dataset as well as write access. The durable spool does not cover this path. `binary::FrameWriter` is public for producers that build frames themselves. diff --git a/src/timeseries/binary.rs b/src/timeseries/binary.rs index fee9834..45cbfac 100644 --- a/src/timeseries/binary.rs +++ b/src/timeseries/binary.rs @@ -743,11 +743,18 @@ fn unprocessable(message: String) -> ResponseError { /// The server rejected the request because the series named in a frame no longer match what the /// client resolved: unknown after a delete, or renamed since. +/// +/// Read from the problem document, not the body text: the SDK's own pre-flight 404 names the +/// series it could not resolve, so an external id spelling one of these slugs used to earn a +/// pointless second attempt. fn is_stale_series_rejection(error: &ResponseError) -> bool { - let status = error.get_status(); - (status == StatusCode::NOT_FOUND || status == StatusCode::UNPROCESSABLE_ENTITY) - && (error.message.contains("unknown-timeseries") - || error.message.contains("external-id-mismatch")) + let Some(problem) = error.problem() else { + return false; + }; + matches!( + problem.slug(), + Some("unknown-timeseries" | "external-id-mismatch") + ) } impl TimeSeriesService { @@ -966,6 +973,47 @@ mod tests { (batches, schema) } + #[test] + fn only_a_stale_series_problem_rebuilds_the_request() { + let refusal = |status: StatusCode, slug: &str, reason: &str| ResponseError { + status, + message: format!( + r#"{{"type":"https://intellistream.ai/errors/{slug}","title":"Binary datapoint request rejected","reason":"{reason}","timeseriesIds":[3]}}"# + ), + content_type: Some("application/problem+json".to_string()), + }; + + assert!(is_stale_series_rejection(&refusal( + StatusCode::NOT_FOUND, + "unknown-timeseries", + "unknown-timeseries" + ))); + assert!(is_stale_series_rejection(&refusal( + StatusCode::UNPROCESSABLE_ENTITY, + "external-id-mismatch", + "external-id-mismatch" + ))); + // Re-resolving cannot make a value fit a type it does not fit. + assert!(!is_stale_series_rejection(&refusal( + StatusCode::UNPROCESSABLE_ENTITY, + "value-type-mismatch", + "value-type-mismatch" + ))); + // The single type every binary refusal carried before platform #120 split it. Deliberately + // not matched: it never reached a release of this SDK, so nothing can be sending it. + assert!(!is_stale_series_rejection(&refusal( + StatusCode::NOT_FOUND, + "datapoint-block-rejected", + "unknown-timeseries" + ))); + // The SDK's own pre-flight 404 is not a problem document, whatever the series is called. + assert!(!is_stale_series_rejection(&ResponseError { + status: StatusCode::NOT_FOUND, + message: "Could not find following timeseries: unknown-timeseries-pump".to_string(), + content_type: None, + })); + } + #[test] fn float_frame_is_sorted_deduplicated_and_readable_by_arrow() { let mut writer = FrameWriter::new(DatapointValueType::Float); diff --git a/src/timeseries/test.rs b/src/timeseries/test.rs index 4512a75..63b8cd6 100644 --- a/src/timeseries/test.rs +++ b/src/timeseries/test.rs @@ -1573,6 +1573,57 @@ mod tests { Ok(()) } + /// Recreating a series under the same external id gives it a new id, so the server refuses the + /// one this service cached — `unknown-timeseries` — and the call must re-resolve and succeed. + /// The only live coverage of that retry: every other path resolves before it sends. Needs a + /// backend that serves the binary endpoint. + #[tokio::test] + #[ignore] + async fn test_insert_datapoints_binary_re_resolves_a_recreated_series() -> Result<(), Box> { + let api_service = create_api_service(); + let ext_id = unique_id("ts_binary_recreated"); + let mut ts_cleanup = cleanup_timeseries(vec![ext_id.clone()]); + + let mut ts_collection = DataWrapper::new(); + ts_collection.add_item( + TimeSeries::builder() + .set_external_id(&ext_id) + .set_name(&ext_id) + .set_unit("celsius") + .set_value_type("float") + .clone(), + ); + let mut data_request: DataWrapper> = DataWrapper::new(); + let mut dp_collection = DatapointsCollection::from_external_id(&ext_id); + dp_collection.datapoints = vec![ + DatapointString::from_datetime(Utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).unwrap(), "42.0"), + ]; + data_request.add_item(dp_collection); + + api_service.time_series.create(&ts_collection).await.expect("could not create the series"); + api_service + .time_series + .insert_datapoints_binary(&data_request, &BinaryIngestOptions::default()) + .await + .expect("the first binary insert failed"); + + delete_timeseries(&api_service, &[&ext_id]).await; + api_service.time_series.create(&ts_collection).await.expect("could not recreate the series"); + + match api_service + .time_series + .insert_datapoints_binary(&data_request, &BinaryIngestOptions::default()) + .await + { + Ok(r) => assert_eq!(r.get_http_status_code().unwrap(), StatusCode::NO_CONTENT.as_u16()), + Err(e) => panic!("the cached id was not re-resolved: {}: {}", e.get_status(), e.get_message()), + } + + delete_timeseries(&api_service, &[&ext_id]).await; + ts_cleanup.disarm(); + Ok(()) + } + fn validate_data_insertion(result: Result, ResponseError>) { match result { Ok(r) => {