gonzalo_store_server/
lib.rs

1//! A `Store` backed by a remote gonzalo daemon. Operators choose the
2//! transport: [`ServerStore::http`] (HTTP/JSON via reqwest) or
3//! [`ServerStore::grpc`] (gRPC via tonic). Both speak the same JSON
4//! serialization of `gonzalo-core` types. Each transport has a
5//! `*_with_token` constructor that sends `Authorization: Bearer <token>`.
6
7use async_trait::async_trait;
8use gonzalo_core::{
9    BlobEntry, BlobStore, ContentHash, CoreError, DeleteResult, Identity, KeyPrefix, PutResult,
10    Record, RecordKey, Result, Revision, Store, store::Conflict,
11};
12use gonzalo_proto::http::{
13    DeleteBody, DeleteOutcome, PurgeBody, PutBody, PutOutcome, RawRecordBody,
14};
15use gonzalo_proto::v1::{
16    DeleteBlobRequest, DeleteRequest, GetBlobRequest, GetRequest, HasBlobRequest, ListBlobsRequest,
17    ListRequest, ListResponse, PurgeRequest, PutBlobRequest, PutRequest, PutResponse,
18    gonzalo_client::GonzaloClient,
19};
20use tonic::transport::Channel;
21
22/// The error every replication call (`get_raw`, `list_raw`, `put_raw`,
23/// `purge`) returns against a daemon without the replication surface: HTTP
24/// `404` on the route, or gRPC `Unimplemented`. There is deliberately **no
25/// fallback** to the consumer routes. Consumer reads hide tombstones and
26/// consumer put re-stamps recreations, so replicating through them resurrects
27/// deleted records (spec §3.6).
28pub const DAEMON_PREDATES_REPLICATION: &str =
29    "daemon predates replication reads (gonzalo#203); upgrade gonzalod";
30
31/// The error `has_blob` returns against a daemon whose gRPC service has no
32/// `HasBlob` RPC (`Unimplemented`).
33///
34/// There is deliberately **no fallback** to `GetBlob`. It would answer
35/// correctly while silently reinstating the per-blob download this check
36/// exists to remove (ADR 0028, gonzalo#329), so a mixed-version deployment
37/// would keep paying the cost with nothing to show why.
38///
39/// HTTP needs no such message: axum serves `HEAD` from a route's `GET`
40/// handler, so an older daemon answers `HEAD /v1/blobs/{hash}` correctly
41/// anyway — it just reads the blob to do it. Only gRPC can see the gap.
42pub const DAEMON_PREDATES_BLOB_PRESENCE: &str =
43    "daemon predates the blob presence check (gonzalo#329); upgrade gonzalod";
44
45enum Backend {
46    Http {
47        base: reqwest::Url,
48        client: reqwest::Client,
49        token: Option<String>,
50    },
51    Grpc {
52        client: GonzaloClient<Channel>,
53        token: Option<String>,
54    },
55}
56
57/// A client substrate over a remote gonzalo daemon.
58pub struct ServerStore {
59    backend: Backend,
60}
61
62impl ServerStore {
63    /// Talk to the daemon's HTTP/JSON API rooted at `base_url`.
64    pub fn http(base_url: &str) -> Result<Self> {
65        Self::http_inner(base_url, None)
66    }
67
68    /// As [`ServerStore::http`], sending a bearer token on every request.
69    pub fn http_with_token(base_url: &str, token: impl Into<String>) -> Result<Self> {
70        Self::http_inner(base_url, Some(token.into()))
71    }
72
73    fn http_inner(base_url: &str, token: Option<String>) -> Result<Self> {
74        let base = reqwest::Url::parse(base_url).map_err(|e| CoreError::Backend(e.to_string()))?;
75        Ok(Self {
76            backend: Backend::Http {
77                base,
78                client: reqwest::Client::new(),
79                token,
80            },
81        })
82    }
83
84    /// Talk to the daemon's gRPC API at `endpoint` (e.g. `http://host:50051`).
85    pub async fn grpc(endpoint: String) -> Result<Self> {
86        Self::grpc_inner(endpoint, None).await
87    }
88
89    /// As [`ServerStore::grpc`], sending a bearer token on every call.
90    pub async fn grpc_with_token(endpoint: String, token: impl Into<String>) -> Result<Self> {
91        Self::grpc_inner(endpoint, Some(token.into())).await
92    }
93
94    async fn grpc_inner(endpoint: String, token: Option<String>) -> Result<Self> {
95        let client = GonzaloClient::connect(endpoint)
96            .await
97            .map_err(|e| CoreError::Backend(e.to_string()))?
98            .max_decoding_message_size(gonzalo_proto::DEFAULT_MAX_BLOB_SIZE);
99        Ok(Self {
100            backend: Backend::Grpc { client, token },
101        })
102    }
103
104    /// `base` + `segments` + the record key's three segments.
105    fn key_url(base: &reqwest::Url, segments: &[&str], key: &RecordKey) -> Result<reqwest::Url> {
106        let mut url = base.clone();
107        url.path_segments_mut()
108            .map_err(|_| CoreError::Backend("base URL cannot be a base".into()))?
109            .extend(segments)
110            .extend([&key.namespace, &key.collection, &key.id]);
111        Ok(url)
112    }
113
114    fn records_url(base: &reqwest::Url, key: &RecordKey) -> Result<reqwest::Url> {
115        Self::key_url(base, &["v1", "records"], key)
116    }
117
118    /// `…/v1/raw/records/{ns}/{col}/{id}` (gonzalo#203).
119    fn raw_records_url(base: &reqwest::Url, key: &RecordKey) -> Result<reqwest::Url> {
120        Self::key_url(base, &["v1", "raw", "records"], key)
121    }
122
123    /// `…/v1/purge/{ns}/{col}/{id}` (gonzalo#203).
124    fn purge_url(base: &reqwest::Url, key: &RecordKey) -> Result<reqwest::Url> {
125        Self::key_url(base, &["v1", "purge"], key)
126    }
127
128    /// `base` + `segments` + `?namespace=&collection=` from `prefix` (shared by
129    /// `/v1/keys` and `/v1/raw/keys`).
130    fn keys_url(
131        base: &reqwest::Url,
132        segments: &[&str],
133        prefix: &KeyPrefix,
134    ) -> Result<reqwest::Url> {
135        let mut url = base.clone();
136        url.path_segments_mut()
137            .map_err(|_| CoreError::Backend("base URL cannot be a base".into()))?
138            .extend(segments);
139        {
140            let mut q = url.query_pairs_mut();
141            if let Some(ns) = &prefix.namespace {
142                q.append_pair("namespace", ns);
143            }
144            if let Some(col) = &prefix.collection {
145                q.append_pair("collection", col);
146            }
147        }
148        Ok(url)
149    }
150
151    /// `…/v1/blobs` (list) or `…/v1/blobs/{hash}` (one blob) when `hash` is set.
152    fn blobs_url(base: &reqwest::Url, hash: Option<&str>) -> Result<reqwest::Url> {
153        let mut url = base.clone();
154        {
155            let mut seg = url
156                .path_segments_mut()
157                .map_err(|_| CoreError::Backend("base URL cannot be a base".into()))?;
158            seg.extend(["v1", "blobs"]);
159            if let Some(h) = hash {
160                seg.push(h);
161            }
162        }
163        Ok(url)
164    }
165}
166
167/// Wrap a gRPC message in a request, attaching the bearer token if present.
168fn grpc_request<T>(msg: T, token: &Option<String>) -> Result<tonic::Request<T>> {
169    let mut req = tonic::Request::new(msg);
170    if let Some(t) = token {
171        let value = format!("Bearer {t}")
172            .parse()
173            .map_err(|_| CoreError::Backend("invalid token characters".into()))?;
174        req.metadata_mut().insert("authorization", value);
175    }
176    Ok(req)
177}
178
179fn maybe_auth(rb: reqwest::RequestBuilder, token: &Option<String>) -> reqwest::RequestBuilder {
180    match token {
181        Some(t) => rb.bearer_auth(t),
182        None => rb,
183    }
184}
185
186#[async_trait]
187impl Store for ServerStore {
188    async fn get(&self, key: &RecordKey) -> Result<Option<Record>> {
189        match &self.backend {
190            Backend::Http {
191                base,
192                client,
193                token,
194            } => {
195                let url = Self::records_url(base, key)?;
196                let resp = maybe_auth(client.get(url), token)
197                    .send()
198                    .await
199                    .map_err(be)?;
200                if resp.status() == reqwest::StatusCode::NOT_FOUND {
201                    return Ok(None);
202                }
203                let resp = ensure_read_ok(resp).await?;
204                Ok(Some(resp.json::<Record>().await.map_err(be)?))
205            }
206            Backend::Grpc { client, token } => {
207                let mut client = client.clone();
208                let req = grpc_request(
209                    GetRequest {
210                        namespace: key.namespace.clone(),
211                        collection: key.collection.clone(),
212                        id: key.id.clone(),
213                    },
214                    token,
215                )?;
216                let resp = client.get(req).await.map_err(status)?.into_inner();
217                decode_get_response(resp)
218            }
219        }
220    }
221
222    async fn put(&self, record: Record, expected: Option<Revision>) -> Result<PutResult> {
223        let key = record.key.clone();
224        match &self.backend {
225            Backend::Http {
226                base,
227                client,
228                token,
229            } => {
230                let url = Self::records_url(base, &key)?;
231                let body = PutBody { record, expected };
232                let resp = maybe_auth(client.put(url).json(&body), token)
233                    .send()
234                    .await
235                    .map_err(be)?;
236                // Read the body as text first so error statuses (403/413/400,
237                // which carry plain-text bodies) surface their real status and
238                // message instead of being masked as a JSON decode error (#147).
239                let status = resp.status();
240                let text = resp.text().await.map_err(be)?;
241                classify_put_response_for(&key, status, &text)
242            }
243            Backend::Grpc { client, token } => {
244                let mut client = client.clone();
245                let req = grpc_request(
246                    PutRequest {
247                        record_json: serde_json::to_vec(&record).map_err(se)?,
248                        expected_json: serde_json::to_vec(&expected).map_err(se)?,
249                    },
250                    token,
251                )?;
252                let resp = client
253                    .put(req)
254                    .await
255                    .map_err(|s| put_status(s, &key))?
256                    .into_inner();
257                decode_put_response(resp)
258            }
259        }
260    }
261
262    async fn list(&self, prefix: &KeyPrefix) -> Result<Vec<RecordKey>> {
263        match &self.backend {
264            Backend::Http {
265                base,
266                client,
267                token,
268            } => {
269                let url = Self::keys_url(base, &["v1", "keys"], prefix)?;
270                let resp = maybe_auth(client.get(url), token)
271                    .send()
272                    .await
273                    .map_err(be)?;
274                let resp = ensure_read_ok(resp).await?;
275                Ok(resp.json::<Vec<RecordKey>>().await.map_err(be)?)
276            }
277            Backend::Grpc { client, token } => {
278                let mut client = client.clone();
279                let req = grpc_request(
280                    ListRequest {
281                        namespace: prefix.namespace.clone(),
282                        collection: prefix.collection.clone(),
283                    },
284                    token,
285                )?;
286                let resp = client.list(req).await.map_err(status)?.into_inner();
287                decode_keys(resp)
288            }
289        }
290    }
291
292    /// Sends the delete with its author. The daemon honours a named deleter
293    /// only for an admin credential or open mode; a non-admin token is always
294    /// stamped as itself (ADR 0015, gonzalo#203).
295    async fn delete_as(
296        &self,
297        key: &RecordKey,
298        expected: Option<Revision>,
299        author: Option<Identity>,
300    ) -> Result<DeleteResult> {
301        match &self.backend {
302            Backend::Http {
303                base,
304                client,
305                token,
306            } => {
307                let url = Self::records_url(base, key)?;
308                let body = DeleteBody { expected, author };
309                let resp = maybe_auth(client.delete(url).json(&body), token)
310                    .send()
311                    .await
312                    .map_err(be)?;
313                // Read the body as text first so error statuses (403/400, which
314                // carry plain-text bodies) surface their real status and message
315                // instead of being masked as a JSON decode error (mirrors `put`).
316                let status = resp.status();
317                let text = resp.text().await.map_err(be)?;
318                classify_delete_response(status, &text)
319            }
320            Backend::Grpc { client, token } => {
321                let mut client = client.clone();
322                let author_json = match &author {
323                    None => Vec::new(),
324                    Some(a) => serde_json::to_vec(a).map_err(se)?,
325                };
326                let req = grpc_request(
327                    DeleteRequest {
328                        namespace: key.namespace.clone(),
329                        collection: key.collection.clone(),
330                        id: key.id.clone(),
331                        expected_json: serde_json::to_vec(&expected).map_err(se)?,
332                        author_json,
333                    },
334                    token,
335                )?;
336                let resp = client.delete(req).await.map_err(status)?.into_inner();
337                decode_delete_outcome(&resp.outcome, &resp.payload_json)
338            }
339        }
340    }
341
342    async fn get_raw(&self, key: &RecordKey) -> Result<Option<Record>> {
343        match &self.backend {
344            Backend::Http {
345                base,
346                client,
347                token,
348            } => {
349                let url = Self::raw_records_url(base, key)?;
350                let resp = maybe_auth(client.get(url), token)
351                    .send()
352                    .await
353                    .map_err(be)?;
354                let status = resp.status();
355                let text = resp.text().await.map_err(be)?;
356                classify_raw_get_response(status, &text)
357            }
358            Backend::Grpc { client, token } => {
359                let mut client = client.clone();
360                let req = grpc_request(
361                    GetRequest {
362                        namespace: key.namespace.clone(),
363                        collection: key.collection.clone(),
364                        id: key.id.clone(),
365                    },
366                    token,
367                )?;
368                let resp = client
369                    .get_raw(req)
370                    .await
371                    .map_err(replication_status)?
372                    .into_inner();
373                decode_get_response(resp)
374            }
375        }
376    }
377
378    async fn list_raw(&self, prefix: &KeyPrefix) -> Result<Vec<RecordKey>> {
379        match &self.backend {
380            Backend::Http {
381                base,
382                client,
383                token,
384            } => {
385                let url = Self::keys_url(base, &["v1", "raw", "keys"], prefix)?;
386                let resp = maybe_auth(client.get(url), token)
387                    .send()
388                    .await
389                    .map_err(be)?;
390                let status = resp.status();
391                let text = resp.text().await.map_err(be)?;
392                classify_raw_list_response(status, &text)
393            }
394            Backend::Grpc { client, token } => {
395                let mut client = client.clone();
396                let req = grpc_request(
397                    ListRequest {
398                        namespace: prefix.namespace.clone(),
399                        collection: prefix.collection.clone(),
400                    },
401                    token,
402                )?;
403                let resp = client
404                    .list_raw(req)
405                    .await
406                    .map_err(replication_status)?
407                    .into_inner();
408                decode_keys(resp)
409            }
410        }
411    }
412
413    async fn put_raw(&self, record: Record, expected: Option<Revision>) -> Result<PutResult> {
414        let key = record.key.clone();
415        match &self.backend {
416            Backend::Http {
417                base,
418                client,
419                token,
420            } => {
421                let url = Self::raw_records_url(base, &key)?;
422                let body = PutBody { record, expected };
423                let resp = maybe_auth(client.put(url).json(&body), token)
424                    .send()
425                    .await
426                    .map_err(be)?;
427                let status = resp.status();
428                let text = resp.text().await.map_err(be)?;
429                classify_raw_put_response(&key, status, &text)
430            }
431            Backend::Grpc { client, token } => {
432                let mut client = client.clone();
433                let req = grpc_request(
434                    PutRequest {
435                        record_json: serde_json::to_vec(&record).map_err(se)?,
436                        expected_json: serde_json::to_vec(&expected).map_err(se)?,
437                    },
438                    token,
439                )?;
440                let resp = client
441                    .put_raw(req)
442                    .await
443                    .map_err(|s| raw_put_status(s, &key))?
444                    .into_inner();
445                decode_put_response(resp)
446            }
447        }
448    }
449
450    async fn purge(&self, key: &RecordKey, expected: Revision) -> Result<DeleteResult> {
451        match &self.backend {
452            Backend::Http {
453                base,
454                client,
455                token,
456            } => {
457                let url = Self::purge_url(base, key)?;
458                let body = PurgeBody { expected };
459                let resp = maybe_auth(client.post(url).json(&body), token)
460                    .send()
461                    .await
462                    .map_err(be)?;
463                let status = resp.status();
464                let text = resp.text().await.map_err(be)?;
465                classify_purge_response(status, &text)
466            }
467            Backend::Grpc { client, token } => {
468                let mut client = client.clone();
469                let req = grpc_request(
470                    PurgeRequest {
471                        namespace: key.namespace.clone(),
472                        collection: key.collection.clone(),
473                        id: key.id.clone(),
474                        expected_json: serde_json::to_vec(&expected).map_err(se)?,
475                    },
476                    token,
477                )?;
478                let resp = client
479                    .purge(req)
480                    .await
481                    .map_err(replication_status)?
482                    .into_inner();
483                decode_delete_outcome(&resp.outcome, &resp.payload_json)
484            }
485        }
486    }
487}
488
489#[async_trait]
490impl BlobStore for ServerStore {
491    async fn put_blob(&self, content: &[u8]) -> Result<ContentHash> {
492        // The blob is content-addressed: compute the hash locally to address
493        // the request, exactly as the daemon will recompute and verify it.
494        let hash = ContentHash::of(content);
495        match &self.backend {
496            Backend::Http {
497                base,
498                client,
499                token,
500            } => {
501                let url = Self::blobs_url(base, Some(&hash.0))?;
502                let resp = maybe_auth(client.put(url).body(content.to_vec()), token)
503                    .send()
504                    .await
505                    .map_err(be)?;
506                let status = resp.status();
507                let text = resp.text().await.map_err(be)?;
508                classify_blob_put_response(status, &text, hash)
509            }
510            Backend::Grpc { client, token } => {
511                let mut client = client.clone();
512                let req = grpc_request(
513                    PutBlobRequest {
514                        hash: hash.0.clone(),
515                        content: content.to_vec(),
516                    },
517                    token,
518                )?;
519                let resp = client.put_blob(req).await.map_err(status)?.into_inner();
520                Ok(ContentHash(resp.hash))
521            }
522        }
523    }
524
525    async fn get_blob(&self, hash: &ContentHash) -> Result<Option<Vec<u8>>> {
526        match &self.backend {
527            Backend::Http {
528                base,
529                client,
530                token,
531            } => {
532                let url = Self::blobs_url(base, Some(&hash.0))?;
533                let resp = maybe_auth(client.get(url), token)
534                    .send()
535                    .await
536                    .map_err(be)?;
537                if resp.status() == reqwest::StatusCode::NOT_FOUND {
538                    return Ok(None);
539                }
540                let resp = ensure_read_ok(resp).await?;
541                Ok(Some(resp.bytes().await.map_err(be)?.to_vec()))
542            }
543            Backend::Grpc { client, token } => {
544                let mut client = client.clone();
545                let req = grpc_request(
546                    GetBlobRequest {
547                        hash: hash.0.clone(),
548                    },
549                    token,
550                )?;
551                let resp = client.get_blob(req).await.map_err(status)?.into_inner();
552                Ok(resp.found.then_some(resp.content))
553            }
554        }
555    }
556
557    /// Presence over the wire, without the bytes.
558    ///
559    /// The trait default fetches the blob and discards it, which over a daemon
560    /// means a full download per newly referenced blob on every commit — 600 KB
561    /// a shard for a vector index (ADR 0028, #329). `HEAD /v1/blobs/{hash}` and
562    /// the `HasBlob` RPC answer from the daemon's own presence check instead.
563    ///
564    /// A `404` is `false`, not an error: absence is the answer, not a failure.
565    async fn has_blob(&self, hash: &ContentHash) -> Result<bool> {
566        match &self.backend {
567            Backend::Http {
568                base,
569                client,
570                token,
571            } => {
572                let url = Self::blobs_url(base, Some(&hash.0))?;
573                let resp = maybe_auth(client.head(url), token)
574                    .send()
575                    .await
576                    .map_err(be)?;
577                if resp.status() == reqwest::StatusCode::NOT_FOUND {
578                    return Ok(false);
579                }
580                ensure_read_ok(resp).await?;
581                Ok(true)
582            }
583            Backend::Grpc { client, token } => {
584                let mut client = client.clone();
585                let req = grpc_request(
586                    HasBlobRequest {
587                        hash: hash.0.clone(),
588                    },
589                    token,
590                )?;
591                Ok(client
592                    .has_blob(req)
593                    .await
594                    .map_err(presence_status)?
595                    .into_inner()
596                    .found)
597            }
598        }
599    }
600
601    async fn list_blobs(&self) -> Result<Vec<BlobEntry>> {
602        match &self.backend {
603            Backend::Http {
604                base,
605                client,
606                token,
607            } => {
608                let url = Self::blobs_url(base, None)?;
609                let resp = maybe_auth(client.get(url), token)
610                    .send()
611                    .await
612                    .map_err(be)?;
613                let resp = ensure_read_ok(resp).await?;
614                Ok(resp.json::<Vec<BlobEntry>>().await.map_err(be)?)
615            }
616            Backend::Grpc { client, token } => {
617                let mut client = client.clone();
618                let req = grpc_request(ListBlobsRequest {}, token)?;
619                let resp = client.list_blobs(req).await.map_err(status)?.into_inner();
620                Ok(resp
621                    .entries
622                    .into_iter()
623                    .map(|e| BlobEntry {
624                        hash: ContentHash(e.hash),
625                        modified_unix_ms: e.modified_unix_ms,
626                    })
627                    .collect())
628            }
629        }
630    }
631
632    async fn delete_blob(&self, hash: &ContentHash) -> Result<()> {
633        match &self.backend {
634            Backend::Http {
635                base,
636                client,
637                token,
638            } => {
639                let url = Self::blobs_url(base, Some(&hash.0))?;
640                let resp = maybe_auth(client.delete(url), token)
641                    .send()
642                    .await
643                    .map_err(be)?;
644                let status = resp.status();
645                if status == reqwest::StatusCode::OK {
646                    return Ok(());
647                }
648                let text = resp.text().await.map_err(be)?;
649                Err(CoreError::Backend(format!(
650                    "daemon returned {status}: {text}"
651                )))
652            }
653            Backend::Grpc { client, token } => {
654                let mut client = client.clone();
655                let req = grpc_request(
656                    DeleteBlobRequest {
657                        hash: hash.0.clone(),
658                    },
659                    token,
660                )?;
661                client.delete_blob(req).await.map_err(status)?;
662                Ok(())
663            }
664        }
665    }
666}
667
668fn delete_outcome_to_result(outcome: DeleteOutcome) -> DeleteResult {
669    match outcome {
670        DeleteOutcome::Deleted => DeleteResult::Deleted,
671        DeleteOutcome::Conflict { conflict } => DeleteResult::Conflict(conflict),
672    }
673}
674
675/// Decide a `delete`'s result from the HTTP response status and body text.
676///
677/// The daemon speaks JSON (`DeleteOutcome`) only for `200 OK` (deleted) and
678/// `409 Conflict`; every other status (`403` authorization denial, `400` bad
679/// request) carries a plain-text body, surfaced verbatim as
680/// `CoreError::Backend("daemon returned <status>: <body>")` (mirrors `put`).
681fn classify_delete_response(status: reqwest::StatusCode, body: &str) -> Result<DeleteResult> {
682    match status {
683        reqwest::StatusCode::OK | reqwest::StatusCode::CONFLICT => {
684            let outcome: DeleteOutcome = serde_json::from_str(body).map_err(se)?;
685            Ok(delete_outcome_to_result(outcome))
686        }
687        other => Err(CoreError::Backend(format!(
688            "daemon returned {other}: {body}"
689        ))),
690    }
691}
692
693fn outcome_to_result(outcome: PutOutcome) -> PutResult {
694    match outcome {
695        PutOutcome::Committed { revision } => PutResult::Committed(revision),
696        PutOutcome::Conflict { conflict } => PutResult::Conflict(conflict),
697    }
698}
699
700/// Decide a `put`'s result from the HTTP response status and body text.
701///
702/// The daemon speaks JSON (`PutOutcome`) only for `200 OK` (committed) and
703/// `409 Conflict`; every other status — notably `403` (namespace authorization
704/// denial), `413` (too large), and `400` (bad request) — carries a plain-text
705/// body. Feeding those bodies to a JSON parser masks the real failure behind a
706/// generic "error decoding response body" (#147), so any other status is
707/// surfaced verbatim as `CoreError::Backend("daemon returned <status>: <body>")`.
708fn classify_put_response(status: reqwest::StatusCode, body: &str) -> Result<PutResult> {
709    match status {
710        reqwest::StatusCode::OK | reqwest::StatusCode::CONFLICT => {
711            let outcome: PutOutcome = serde_json::from_str(body).map_err(se)?;
712            Ok(outcome_to_result(outcome))
713        }
714        // A `400` is always the call being wrong — a consumer put of a
715        // tombstone, or a path/body key disagreement (#158) — never the store
716        // failing. Restore it as `Invalid` so a store behind the daemon fails
717        // the way a local one does instead of looking like an outage, keeping
718        // the status in the message as every other failure does (#299).
719        reqwest::StatusCode::BAD_REQUEST => Err(CoreError::Invalid(format!(
720            "daemon returned {}: {body}",
721            reqwest::StatusCode::BAD_REQUEST
722        ))),
723        other => Err(CoreError::Backend(format!(
724            "daemon returned {other}: {body}"
725        ))),
726    }
727}
728
729/// [`classify_put_response`], plus `412 Precondition Failed`: the daemon's
730/// `CoreError::NotFound` (`expected` names a revision the store does not hold),
731/// restored as `NotFound(key)`.
732fn classify_put_response_for(
733    key: &RecordKey,
734    status: reqwest::StatusCode,
735    body: &str,
736) -> Result<PutResult> {
737    match status {
738        reqwest::StatusCode::PRECONDITION_FAILED => Err(CoreError::NotFound(key.clone())),
739        other => classify_put_response(other, body),
740    }
741}
742
743/// A raw put's result: `404` means the route does not exist →
744/// [`upgrade_required`], never a fallback; otherwise as a consumer put.
745fn classify_raw_put_response(
746    key: &RecordKey,
747    status: reqwest::StatusCode,
748    body: &str,
749) -> Result<PutResult> {
750    match status {
751        reqwest::StatusCode::NOT_FOUND => Err(upgrade_required()),
752        other => classify_put_response_for(key, other, body),
753    }
754}
755
756/// Decode a gRPC `PutResponse` (shared by `Put` and `PutRaw`).
757fn decode_put_response(resp: PutResponse) -> Result<PutResult> {
758    match resp.outcome.as_str() {
759        "committed" => {
760            let rev: Revision = serde_json::from_slice(&resp.payload_json).map_err(se)?;
761            Ok(PutResult::Committed(rev))
762        }
763        "conflict" => {
764            let c: Conflict = serde_json::from_slice(&resp.payload_json).map_err(se)?;
765            Ok(PutResult::Conflict(Box::new(c)))
766        }
767        other => Err(CoreError::Backend(format!("unknown put outcome: {other}"))),
768    }
769}
770
771/// Decode a gRPC `GetResponse` (shared by `Get` and `GetRaw`).
772fn decode_get_response(resp: gonzalo_proto::v1::GetResponse) -> Result<Option<Record>> {
773    if resp.found {
774        Ok(Some(serde_json::from_slice(&resp.record_json).map_err(se)?))
775    } else {
776        Ok(None)
777    }
778}
779
780/// Decode a gRPC delete-shaped outcome (shared by `Delete` and `Purge`).
781fn decode_delete_outcome(outcome: &str, payload_json: &[u8]) -> Result<DeleteResult> {
782    match outcome {
783        "deleted" => Ok(DeleteResult::Deleted),
784        "conflict" => {
785            let c: Conflict = serde_json::from_slice(payload_json).map_err(se)?;
786            Ok(DeleteResult::Conflict(Box::new(c)))
787        }
788        other => Err(CoreError::Backend(format!(
789            "unknown delete outcome: {other}"
790        ))),
791    }
792}
793
794/// [`DAEMON_PREDATES_REPLICATION`] as a `CoreError`.
795fn upgrade_required() -> CoreError {
796    CoreError::Backend(DAEMON_PREDATES_REPLICATION.into())
797}
798
799/// Decide a raw `get` from the HTTP status and body text. `200` carries a
800/// [`RawRecordBody`] (absence is `{"record": null}`); `404` →
801/// [`upgrade_required`]; any other status surfaces the daemon's body (#195).
802fn classify_raw_get_response(status: reqwest::StatusCode, body: &str) -> Result<Option<Record>> {
803    match status {
804        reqwest::StatusCode::OK => Ok(serde_json::from_str::<RawRecordBody>(body)
805            .map_err(se)?
806            .record),
807        reqwest::StatusCode::NOT_FOUND => Err(upgrade_required()),
808        other => Err(read_response_error(other, body)),
809    }
810}
811
812/// Decide a raw `list`: `200` → keys, `404` → [`upgrade_required`], otherwise
813/// the daemon's body.
814fn classify_raw_list_response(status: reqwest::StatusCode, body: &str) -> Result<Vec<RecordKey>> {
815    match status {
816        reqwest::StatusCode::OK => serde_json::from_str(body).map_err(se),
817        reqwest::StatusCode::NOT_FOUND => Err(upgrade_required()),
818        other => Err(read_response_error(other, body)),
819    }
820}
821
822/// Decide a `purge`: `200`/`409` carry a [`DeleteOutcome`], `404` →
823/// [`upgrade_required`], and any other status (`403` non-admin, `400` bad body)
824/// surfaces verbatim, as `delete` does.
825fn classify_purge_response(status: reqwest::StatusCode, body: &str) -> Result<DeleteResult> {
826    match status {
827        reqwest::StatusCode::NOT_FOUND => Err(upgrade_required()),
828        other => classify_delete_response(other, body),
829    }
830}
831
832/// Map a gRPC failure on a replication RPC: `Unimplemented` means the daemon
833/// has no such RPC → [`upgrade_required`]; anything else maps as usual.
834fn replication_status(s: tonic::Status) -> CoreError {
835    if s.code() == tonic::Code::Unimplemented {
836        upgrade_required()
837    } else {
838        status(s)
839    }
840}
841
842/// Map a gRPC failure on `HasBlob`: `Unimplemented` means the daemon predates
843/// the presence check → [`DAEMON_PREDATES_BLOB_PRESENCE`].
844fn presence_status(s: tonic::Status) -> CoreError {
845    if s.code() == tonic::Code::Unimplemented {
846        CoreError::Backend(DAEMON_PREDATES_BLOB_PRESENCE.into())
847    } else {
848        status(s)
849    }
850}
851
852/// Map a gRPC failure on `Put`: `FailedPrecondition` is the daemon's
853/// `CoreError::NotFound`, restored as `NotFound(key)`.
854fn put_status(s: tonic::Status, key: &RecordKey) -> CoreError {
855    if s.code() == tonic::Code::FailedPrecondition {
856        CoreError::NotFound(key.clone())
857    } else if s.code() == tonic::Code::InvalidArgument {
858        // The daemon rejected the call itself; mirrors HTTP's 400 so both
859        // transports surface `Invalid` (gonzalo#299).
860        CoreError::Invalid(s.message().to_string())
861    } else {
862        status(s)
863    }
864}
865
866/// Map a gRPC failure on `PutRaw`: `Unimplemented` → [`upgrade_required`],
867/// otherwise as [`put_status`].
868fn raw_put_status(s: tonic::Status, key: &RecordKey) -> CoreError {
869    if s.code() == tonic::Code::Unimplemented {
870        upgrade_required()
871    } else {
872        put_status(s, key)
873    }
874}
875
876/// Decode a `ListResponse`'s JSON keys (shared by `List` and `ListRaw`).
877fn decode_keys(resp: ListResponse) -> Result<Vec<RecordKey>> {
878    resp.keys_json
879        .iter()
880        .map(|b| serde_json::from_slice::<RecordKey>(b).map_err(se))
881        .collect()
882}
883
884/// Decide a blob `put`'s result from the HTTP response status and body text.
885/// `200 OK` → the (already-known) `hash`; every other status carries a plain
886/// text body surfaced verbatim as `Backend("daemon returned <status>: <body>")`
887/// — notably `413` (too large), `403` (authz), and `400` (hash mismatch) — so
888/// the real failure is never masked (#147).
889fn classify_blob_put_response(
890    status: reqwest::StatusCode,
891    body: &str,
892    hash: ContentHash,
893) -> Result<ContentHash> {
894    match status {
895        reqwest::StatusCode::OK => Ok(hash),
896        other => Err(CoreError::Backend(format!(
897            "daemon returned {other}: {body}"
898        ))),
899    }
900}
901
902/// The error a non-success **read** response carries.
903///
904/// The write paths surface a non-success status together with the daemon's
905/// plain-text body (#147), so a `403` authz denial or a `413` says why. The
906/// reads used reqwest's `error_for_status()`, which **discards the body** —
907/// leaving a generic "HTTP status client error … for url …" and making a failed
908/// remote read strictly harder to debug than a failed write. Same format both
909/// ways now (#195).
910fn read_response_error(status: reqwest::StatusCode, body: &str) -> CoreError {
911    CoreError::Backend(format!("daemon returned {status}: {body}"))
912}
913
914/// Pass `resp` through when the daemon answered successfully; otherwise consume
915/// its body and fail with [`read_response_error`].
916///
917/// `404` never reaches here for `get`/`get_blob` — absence is `Ok(None)` and
918/// those call sites check it first. For `list`/`list_blobs` a `404` is a real
919/// error, since the collection endpoint always exists.
920async fn ensure_read_ok(resp: reqwest::Response) -> Result<reqwest::Response> {
921    let status = resp.status();
922    if status.is_success() {
923        // Hand it back untouched: only the caller knows how to decode it.
924        return Ok(resp);
925    }
926    let body = resp.text().await.unwrap_or_default();
927    Err(read_response_error(status, &body))
928}
929
930fn be<E: std::fmt::Display>(e: E) -> CoreError {
931    CoreError::Backend(e.to_string())
932}
933fn se<E: std::fmt::Display>(e: E) -> CoreError {
934    CoreError::Serde(e.to_string())
935}
936fn status(s: tonic::Status) -> CoreError {
937    CoreError::Backend(s.to_string())
938}
939
940#[cfg(test)]
941mod tests {
942    use super::*;
943    use gonzalo_core::store::Conflict;
944    use gonzalo_core::{Body, ContentHash, Identity, Meta, Record, RecordKind};
945    use reqwest::StatusCode;
946    use std::collections::BTreeMap;
947
948    fn sample_record() -> Record {
949        let body = Body::Inline(b"hello".to_vec());
950        Record {
951            key: RecordKey::new("ns", "col", "id"),
952            kind: RecordKind::Topic,
953            revision: Revision::initial(body.bytes()),
954            parent: None,
955            body,
956            meta: Meta {
957                author: Identity::new("tester"),
958                origin_system: "test".into(),
959                created: 0,
960                updated: 0,
961                labels: BTreeMap::new(),
962            },
963            links: Vec::new(),
964            ancestors: Vec::new(),
965            deleted_at: None,
966            deleted_blob: None,
967            deleted_kind: None,
968        }
969    }
970
971    // ── read paths report failures like the write paths do (#195) ───────────
972
973    /// A `403` on a read names the daemon's reason, not reqwest's generic
974    /// "HTTP status client error". This is the whole point of #195.
975    #[test]
976    fn a_read_failure_carries_the_daemon_body() {
977        let err = read_response_error(StatusCode::FORBIDDEN, "namespace not authorized");
978        let CoreError::Backend(msg) = err else {
979            panic!("read failures are Backend errors");
980        };
981        assert!(msg.contains("403"), "{msg}");
982        assert!(msg.contains("namespace not authorized"), "{msg}");
983    }
984
985    /// The harmonization claim itself: for the same status and body, a read
986    /// error reads exactly like the write error it used to differ from. If
987    /// someone changes one format, this fails rather than letting them drift.
988    #[test]
989    fn reads_and_writes_report_a_failure_identically() {
990        let (status, body) = (StatusCode::FORBIDDEN, "namespace not authorized");
991        let read = read_response_error(status, body);
992        let write = classify_put_response(status, body).unwrap_err();
993        let blob_write =
994            classify_blob_put_response(status, body, ContentHash("h".into())).unwrap_err();
995        assert_eq!(read.to_string(), write.to_string());
996        assert_eq!(read.to_string(), blob_write.to_string());
997    }
998
999    /// A `400` is the daemon saying the call itself is wrong (a consumer put of
1000    /// a tombstone). It comes back as `Invalid`, so a daemon-backed store fails
1001    /// the way a local one does instead of looking like an outage (#299).
1002    #[test]
1003    fn a_bad_request_becomes_invalid_not_backend() {
1004        let err = classify_put_response(
1005            StatusCode::BAD_REQUEST,
1006            "consumer put cannot write a tombstone; use delete_as",
1007        )
1008        .unwrap_err();
1009        assert!(
1010            matches!(err, CoreError::Invalid(ref m) if m.contains("delete_as")),
1011            "got {err:?}"
1012        );
1013    }
1014
1015    /// `413` is the other status #147 cared about; it must survive a read too.
1016    #[test]
1017    fn a_read_failure_preserves_any_status() {
1018        let msg =
1019            read_response_error(StatusCode::PAYLOAD_TOO_LARGE, "blob exceeds 64 MiB").to_string();
1020        assert!(msg.contains("413"), "{msg}");
1021        assert!(msg.contains("blob exceeds 64 MiB"), "{msg}");
1022    }
1023
1024    /// An empty body still produces a usable message — the status alone is the
1025    /// signal, and it must not read as a truncated or malformed error.
1026    #[test]
1027    fn a_read_failure_with_no_body_still_names_the_status() {
1028        let msg = read_response_error(StatusCode::BAD_GATEWAY, "").to_string();
1029        assert!(msg.contains("502"), "{msg}");
1030    }
1031
1032    /// `200 OK` carries a `committed` JSON body → `PutResult::Committed`.
1033    #[test]
1034    fn ok_body_parses_committed() {
1035        let revision = Revision::initial(b"hello");
1036        let json = serde_json::to_string(&PutOutcome::Committed {
1037            revision: revision.clone(),
1038        })
1039        .unwrap();
1040        let result = classify_put_response(StatusCode::OK, &json).unwrap();
1041        assert!(matches!(result, PutResult::Committed(r) if r == revision));
1042    }
1043
1044    /// `409 Conflict` carries a `conflict` JSON body → `PutResult::Conflict`.
1045    #[test]
1046    fn conflict_body_parses_conflict() {
1047        let record = sample_record();
1048        let conflict = Conflict {
1049            key: record.key.clone(),
1050            expected: None,
1051            current: record,
1052        };
1053        let json = serde_json::to_string(&PutOutcome::Conflict {
1054            conflict: Box::new(conflict),
1055        })
1056        .unwrap();
1057        let result = classify_put_response(StatusCode::CONFLICT, &json).unwrap();
1058        assert!(matches!(result, PutResult::Conflict(_)));
1059    }
1060
1061    /// #147: a `403` with a plain-text body surfaces the status + body as a
1062    /// `Backend` error — NOT a masked JSON decode error.
1063    #[test]
1064    fn forbidden_surfaces_status_and_body() {
1065        let body = "principal \"alice\" lacks Write on namespace \"secrets\"";
1066        let err = classify_put_response(StatusCode::FORBIDDEN, body).unwrap_err();
1067        match err {
1068            CoreError::Backend(msg) => {
1069                assert!(msg.contains("403"), "want status 403 in {msg:?}");
1070                assert!(msg.contains(body), "want daemon body in {msg:?}");
1071                assert!(
1072                    !msg.contains("decoding"),
1073                    "must not be a decode error: {msg:?}"
1074                );
1075            }
1076            other => panic!("expected Backend error, got {other:?}"),
1077        }
1078    }
1079
1080    /// `413 Payload Too Large` (plain-text body) also surfaces status + body.
1081    #[test]
1082    fn payload_too_large_surfaces_status_and_body() {
1083        let body = "record exceeds max size";
1084        let err = classify_put_response(StatusCode::PAYLOAD_TOO_LARGE, body).unwrap_err();
1085        match err {
1086            CoreError::Backend(msg) => {
1087                assert!(msg.contains("413"), "want status 413 in {msg:?}");
1088                assert!(msg.contains(body), "want daemon body in {msg:?}");
1089            }
1090            other => panic!("expected Backend error, got {other:?}"),
1091        }
1092    }
1093
1094    /// `400 Bad Request` (plain-text body) surfaces status + body, not a decode
1095    /// error — as `Invalid`, since a `400` is always the call being wrong
1096    /// rather than the store failing (#299).
1097    #[test]
1098    fn bad_request_surfaces_status_and_body() {
1099        let body = "path/body key disagreement";
1100        let err = classify_put_response(StatusCode::BAD_REQUEST, body).unwrap_err();
1101        match err {
1102            CoreError::Invalid(msg) => {
1103                assert!(msg.contains("400"), "want status 400 in {msg:?}");
1104                assert!(msg.contains(body), "want daemon body in {msg:?}");
1105            }
1106            other => panic!("expected Invalid error, got {other:?}"),
1107        }
1108    }
1109
1110    #[test]
1111    fn blob_put_ok_returns_the_hash() {
1112        let content = b"blob body";
1113        let hash = ContentHash::of(content);
1114        let result = classify_blob_put_response(StatusCode::OK, "", hash.clone()).unwrap();
1115        assert_eq!(result, hash);
1116    }
1117
1118    #[test]
1119    fn blob_put_413_surfaces_status_and_body() {
1120        let hash = ContentHash::of(b"x");
1121        let err = classify_blob_put_response(
1122            StatusCode::PAYLOAD_TOO_LARGE,
1123            "blob exceeds max size",
1124            hash,
1125        )
1126        .unwrap_err();
1127        match err {
1128            CoreError::Backend(msg) => {
1129                assert!(msg.contains("413"), "want status 413 in {msg:?}");
1130                assert!(
1131                    msg.contains("blob exceeds max size"),
1132                    "want body in {msg:?}"
1133                );
1134            }
1135            other => panic!("expected Backend error, got {other:?}"),
1136        }
1137    }
1138
1139    #[test]
1140    fn blob_put_400_mismatch_surfaces_status_and_body() {
1141        let hash = ContentHash::of(b"x");
1142        let err = classify_blob_put_response(
1143            StatusCode::BAD_REQUEST,
1144            "blob content does not match the URL hash",
1145            hash,
1146        )
1147        .unwrap_err();
1148        match err {
1149            CoreError::Backend(msg) => {
1150                assert!(msg.contains("400"), "want status 400 in {msg:?}");
1151                assert!(msg.contains("does not match"), "want body in {msg:?}");
1152            }
1153            other => panic!("expected Backend error, got {other:?}"),
1154        }
1155    }
1156
1157    // ── replication surface: upgrade error, never a fallback (#203) ─────────
1158
1159    fn upgrade_error() -> String {
1160        CoreError::Backend(DAEMON_PREDATES_REPLICATION.into()).to_string()
1161    }
1162
1163    #[test]
1164    fn raw_get_200_null_is_absent_and_200_record_is_present() {
1165        assert_eq!(
1166            classify_raw_get_response(StatusCode::OK, r#"{"record":null}"#).unwrap(),
1167            None
1168        );
1169        let rec = sample_record();
1170        let body = serde_json::to_string(&RawRecordBody {
1171            record: Some(rec.clone()),
1172        })
1173        .unwrap();
1174        assert_eq!(
1175            classify_raw_get_response(StatusCode::OK, &body).unwrap(),
1176            Some(rec)
1177        );
1178    }
1179
1180    #[test]
1181    fn replication_404_is_the_upgrade_error() {
1182        let key = RecordKey::new("ns", "col", "id");
1183        for err in [
1184            classify_raw_get_response(StatusCode::NOT_FOUND, "")
1185                .unwrap_err()
1186                .to_string(),
1187            classify_raw_list_response(StatusCode::NOT_FOUND, "")
1188                .unwrap_err()
1189                .to_string(),
1190            classify_raw_put_response(&key, StatusCode::NOT_FOUND, "")
1191                .unwrap_err()
1192                .to_string(),
1193            classify_purge_response(StatusCode::NOT_FOUND, "")
1194                .unwrap_err()
1195                .to_string(),
1196        ] {
1197            assert_eq!(err, upgrade_error());
1198        }
1199    }
1200
1201    #[test]
1202    fn put_412_is_not_found_for_the_record_key() {
1203        let key = RecordKey::new("ns", "col", "id");
1204        assert!(matches!(
1205            classify_put_response_for(&key, StatusCode::PRECONDITION_FAILED, "record not found"),
1206            Err(CoreError::NotFound(k)) if k == key
1207        ));
1208        assert!(matches!(
1209            classify_raw_put_response(&key, StatusCode::PRECONDITION_FAILED, "record not found"),
1210            Err(CoreError::NotFound(k)) if k == key
1211        ));
1212        // The consumer put route never turns a 404 into the upgrade error.
1213        let msg = classify_put_response_for(&key, StatusCode::NOT_FOUND, "nope")
1214            .unwrap_err()
1215            .to_string();
1216        assert_ne!(msg, upgrade_error());
1217    }
1218
1219    #[test]
1220    fn replication_403_keeps_the_daemon_body() {
1221        let body = "principal \"w\" is not an admin; purge requires admin";
1222        let msg = classify_purge_response(StatusCode::FORBIDDEN, body)
1223            .unwrap_err()
1224            .to_string();
1225        assert!(msg.contains("403") && msg.contains(body), "{msg}");
1226        let msg = classify_raw_list_response(StatusCode::FORBIDDEN, "nope")
1227            .unwrap_err()
1228            .to_string();
1229        assert!(msg.contains("403") && msg.contains("nope"), "{msg}");
1230    }
1231
1232    #[test]
1233    fn purge_200_and_409_parse_delete_outcomes() {
1234        let ok = serde_json::to_string(&DeleteOutcome::Deleted).unwrap();
1235        assert_eq!(
1236            classify_purge_response(StatusCode::OK, &ok).unwrap(),
1237            DeleteResult::Deleted
1238        );
1239        let rec = sample_record();
1240        let conflict = serde_json::to_string(&DeleteOutcome::Conflict {
1241            conflict: Box::new(Conflict {
1242                key: rec.key.clone(),
1243                expected: None,
1244                current: rec,
1245            }),
1246        })
1247        .unwrap();
1248        assert!(matches!(
1249            classify_purge_response(StatusCode::CONFLICT, &conflict).unwrap(),
1250            DeleteResult::Conflict(_)
1251        ));
1252    }
1253
1254    #[test]
1255    fn grpc_status_mapping_for_replication_and_puts() {
1256        let key = RecordKey::new("ns", "col", "id");
1257        assert_eq!(
1258            replication_status(tonic::Status::unimplemented("GetRaw")).to_string(),
1259            upgrade_error()
1260        );
1261        assert_ne!(
1262            replication_status(tonic::Status::permission_denied("no")).to_string(),
1263            upgrade_error()
1264        );
1265        assert!(matches!(
1266            put_status(tonic::Status::failed_precondition("record not found"), &key),
1267            CoreError::NotFound(k) if k == key
1268        ));
1269        assert_eq!(
1270            raw_put_status(tonic::Status::unimplemented("PutRaw"), &key).to_string(),
1271            upgrade_error()
1272        );
1273        assert!(matches!(
1274            raw_put_status(tonic::Status::failed_precondition("x"), &key),
1275            CoreError::NotFound(_)
1276        ));
1277    }
1278
1279    /// Spec §6.5: against a daemon without the replication routes, every
1280    /// replication call fails with the upgrade error and **no consumer route is
1281    /// ever hit**.
1282    #[tokio::test]
1283    async fn http_old_daemon_errors_and_never_falls_back_to_consumer_routes() {
1284        use wiremock::matchers::path_regex;
1285        use wiremock::{Mock, MockServer, ResponseTemplate};
1286
1287        let server = MockServer::start().await;
1288        // Any consumer-route hit is a fallback: verification fails the test.
1289        Mock::given(path_regex(r"^/v1/(records|keys)(/|$)"))
1290            .respond_with(ResponseTemplate::new(200))
1291            .named("consumer route (fallback)")
1292            .expect(0)
1293            .mount(&server)
1294            .await;
1295        // An old daemon has no replication routes.
1296        Mock::given(path_regex(r"^/v1/(raw|purge)/"))
1297            .respond_with(ResponseTemplate::new(404))
1298            .named("replication route")
1299            .expect(4)
1300            .mount(&server)
1301            .await;
1302
1303        let store = ServerStore::http(&server.uri()).unwrap();
1304        let key = RecordKey::new("ns", "col", "id");
1305        let errors = [
1306            store.get_raw(&key).await.unwrap_err(),
1307            store.list_raw(&KeyPrefix::default()).await.unwrap_err(),
1308            store.put_raw(sample_record(), None).await.unwrap_err(),
1309            store
1310                .purge(&key, Revision::initial(b"x"))
1311                .await
1312                .unwrap_err(),
1313        ];
1314        for err in errors {
1315            assert_eq!(err.to_string(), upgrade_error());
1316        }
1317        server.verify().await;
1318    }
1319
1320    /// A daemon predating `HasBlob` answers it `Unimplemented`.
1321    ///
1322    /// Falling back to a download would answer correctly while quietly
1323    /// reinstating the per-blob cost this check exists to remove — an
1324    /// invisible regression on a mixed-version deployment. So the client says
1325    /// what to do instead, as it does for the replication RPCs (#329).
1326    #[tokio::test]
1327    async fn grpc_old_daemon_has_blob_asks_for_an_upgrade() {
1328        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1329        let addr = listener.local_addr().unwrap();
1330        tokio::spawn(
1331            tonic::transport::Server::builder()
1332                .add_routes(tonic::service::Routes::default())
1333                .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener)),
1334        );
1335        let store = ServerStore::grpc(format!("http://{addr}")).await.unwrap();
1336        let err = store
1337            .has_blob(&ContentHash::of(b"anything"))
1338            .await
1339            .unwrap_err();
1340        assert_eq!(
1341            err.to_string(),
1342            CoreError::Backend(DAEMON_PREDATES_BLOB_PRESENCE.into()).to_string()
1343        );
1344    }
1345
1346    /// A tonic server with no services answers every RPC `Unimplemented`,
1347    /// which is what a pre-#203 daemon does for the replication RPCs.
1348    #[tokio::test]
1349    async fn grpc_old_daemon_errors_with_the_upgrade_message() {
1350        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1351        let addr = listener.local_addr().unwrap();
1352        tokio::spawn(
1353            tonic::transport::Server::builder()
1354                .add_routes(tonic::service::Routes::default())
1355                .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener)),
1356        );
1357        let store = ServerStore::grpc(format!("http://{addr}")).await.unwrap();
1358        let key = RecordKey::new("ns", "col", "id");
1359        let errors = [
1360            store.get_raw(&key).await.unwrap_err(),
1361            store.list_raw(&KeyPrefix::default()).await.unwrap_err(),
1362            store.put_raw(sample_record(), None).await.unwrap_err(),
1363            store
1364                .purge(&key, Revision::initial(b"x"))
1365                .await
1366                .unwrap_err(),
1367        ];
1368        for err in errors {
1369            assert_eq!(err.to_string(), upgrade_error());
1370        }
1371    }
1372
1373    // ── delete_as sends its author (R1, spec §3.6 amended in 4355f4d) ───────
1374
1375    /// A struct-based matcher (wiremock's `Match` trait has no built-in
1376    /// negation) asserting the request body carries no `author` key at all —
1377    /// `DeleteBody`'s `skip_serializing_if` omits it entirely when `None`.
1378    struct NoAuthorField;
1379    impl wiremock::Match for NoAuthorField {
1380        fn matches(&self, request: &wiremock::Request) -> bool {
1381            serde_json::from_slice::<serde_json::Value>(&request.body)
1382                .map(|v| v.get("author").is_none())
1383                .unwrap_or(false)
1384        }
1385    }
1386
1387    #[tokio::test]
1388    async fn http_delete_as_sends_the_author() {
1389        use wiremock::matchers::{body_partial_json, method, path};
1390        use wiremock::{Mock, MockServer, ResponseTemplate};
1391
1392        let server = MockServer::start().await;
1393
1394        // Identity's real serde shape is `{"id": ..., "display": ...}`, not a
1395        // bare string, so match on the nested field (wiremock's partial-JSON
1396        // matcher is inclusive at every level).
1397        Mock::given(method("DELETE"))
1398            .and(path("/v1/records/ns/col/with-author"))
1399            .and(body_partial_json(
1400                serde_json::json!({"author": {"id": "origin"}}),
1401            ))
1402            .respond_with(ResponseTemplate::new(200).set_body_json(&DeleteOutcome::Deleted))
1403            .named("delete with author")
1404            .expect(1)
1405            .mount(&server)
1406            .await;
1407
1408        Mock::given(method("DELETE"))
1409            .and(path("/v1/records/ns/col/no-author"))
1410            .and(NoAuthorField)
1411            .respond_with(ResponseTemplate::new(200).set_body_json(&DeleteOutcome::Deleted))
1412            .named("delete without author")
1413            .expect(1)
1414            .mount(&server)
1415            .await;
1416
1417        let store = ServerStore::http(&server.uri()).unwrap();
1418
1419        let with_key = RecordKey::new("ns", "col", "with-author");
1420        let result = store
1421            .delete_as(&with_key, None, Some(Identity::new("origin")))
1422            .await
1423            .unwrap();
1424        assert_eq!(result, DeleteResult::Deleted);
1425
1426        let no_key = RecordKey::new("ns", "col", "no-author");
1427        let result = store.delete_as(&no_key, None, None).await.unwrap();
1428        assert_eq!(result, DeleteResult::Deleted);
1429
1430        server.verify().await;
1431    }
1432}
gonzalo_store_server/lib.rs

gonzalo_store_server/
lib.rs
