agentplane/store/
postgres_registry.rs

1//! A durable manifest registry on `PostgreSQL`.
2//!
3//! The backend the immutability rule is actually about. On one node almost
4//! anything holds it up; the moment two instances publish concurrently, only
5//! the database can arbitrate — so the rule is enforced by a **unique primary
6//! key plus one transaction that reads and writes**, not by a `SELECT` this
7//! process makes a decision from and then hopes about.
8//!
9//! The decision itself is
10//! [`decide_publish`](crate::manifest::registry::decide_publish), shared with
11//! the embedded backend and with `MemoryRegistry`. Three hand-written copies of
12//! "may this replace what is there" are three chances to get it subtly
13//! different, and the one that is wrong is whichever nobody tested.
14//!
15//! The lock is a **transaction-scoped advisory lock on the key**, taken before
16//! the read. `SELECT … FOR UPDATE` cannot do this job: it locks a row that
17//! exists, and the race that matters is two publishes of different content
18//! both finding *no* row — both decide to insert, and whichever lands second
19//! either fails on the primary key with an error that reads like a backend
20//! fault, or (under an upsert) silently wins. The advisory lock serialises on
21//! the key whether or not the row is there yet, so the second publisher reads
22//! the first one's row and is refused as the immutability rule intends.
23
24use async_trait::async_trait;
25
26use crate::core::{Digest, KeyId, KeySignature, Signer, StoreError, Verifier};
27use crate::manifest::registry::{
28    PublishVerdict, check_signature, decide_publish, prepare_publish, reparse, sign_manifest,
29};
30use crate::manifest::{Manifest, Registry, RegistryError};
31
32use super::postgres::PostgresStore;
33
34fn be(e: &tokio_postgres::Error) -> RegistryError {
35    RegistryError::Backend(super::postgres::detail(e))
36}
37
38fn pool_err(e: &impl std::fmt::Display) -> RegistryError {
39    RegistryError::Backend(e.to_string())
40}
41
42/// The stored signature, if the row carries one.
43///
44/// An empty key id is the absence: a signature cannot exist without one, so
45/// there is no sentinel here overlapping a value it might be confused with.
46fn stored_signature(key_id: &str, signature: &str) -> Option<KeySignature> {
47    if key_id.is_empty() {
48        return None;
49    }
50    Some(KeySignature {
51        key_id: key_id.to_owned(),
52        // A signature that does not decode cannot verify, and an empty one
53        // fails exactly as a wrong one does — `BadSignature` either way.
54        signature: hex::decode(signature).unwrap_or_default(),
55    })
56}
57
58impl PostgresStore {
59    /// Apply one publish: lock, decide and write in one transaction.
60    async fn publish_row(
61        &self,
62        manifest: &Manifest,
63        signature: Option<KeySignature>,
64    ) -> Result<Digest, RegistryError> {
65        let name = manifest.metadata.name.clone();
66        let version = manifest.metadata.version.clone();
67        let (digest, yaml) = prepare_publish(manifest)?;
68        let (key_id, signature_hex) = signature
69            .as_ref()
70            .map_or((String::new(), String::new()), |a| {
71                (a.key_id.clone(), hex::encode(&a.signature))
72            });
73
74        let mut client = self.pool_ref().get().await.map_err(|e| pool_err(&e))?;
75        let tx = client.transaction().await.map_err(|e| be(&e))?;
76        let tenant = self.tenant_name();
77
78        // Serialise on the key before reading, whether or not a row exists.
79        // `hashtext` of the joined key is a 32-bit lock id; a collision between
80        // two unrelated keys only serialises two publishes that did not need
81        // it, never admits one that should have waited.
82        tx.execute(
83            "SELECT pg_advisory_xact_lock(hashtext($1 || E'\\x1f' || $2 || E'\\x1f' || $3))",
84            &[&tenant, &name, &version],
85        )
86        .await
87        .map_err(|e| be(&e))?;
88
89        let existing = tx
90            .query_opt(
91                "SELECT digest, key_id, signature FROM registry_manifests
92                 WHERE tenant = $1 AND name = $2 AND version = $3",
93                &[&tenant, &name, &version],
94            )
95            .await
96            .map_err(|e| be(&e))?;
97
98        let stored = match &existing {
99            Some(row) => {
100                let hex: String = row.get(0);
101                let stored_digest = Digest::from_hex(&hex).map_err(|_| {
102                    RegistryError::Backend(format!(
103                        "registry row '{name}' '{version}' holds '{hex}', which is not a digest"
104                    ))
105                })?;
106                Some((
107                    stored_digest,
108                    stored_signature(&row.get::<_, String>(1), &row.get::<_, String>(2)),
109                ))
110            }
111            None => None,
112        };
113
114        let verdict = decide_publish(
115            &name,
116            &version,
117            digest,
118            signature.as_ref(),
119            stored.as_ref().map(|(d, a)| (*d, a.as_ref())),
120        );
121        // The refusal rolls back rather than committing nothing: an open
122        // transaction returned to the pool is a lock held by whoever gets the
123        // connection next.
124        let verdict = match verdict {
125            Ok(verdict) => verdict,
126            Err(refusal) => {
127                let _ = tx.rollback().await;
128                return Err(refusal);
129            }
130        };
131
132        match verdict {
133            // A plain insert, never an upsert: the advisory lock makes a
134            // conflict here impossible, and an `ON CONFLICT DO UPDATE` would
135            // turn any future hole in that reasoning into a silent overwrite —
136            // the exact outcome the primary key exists to refuse.
137            PublishVerdict::Insert => {
138                tx.execute(
139                    "INSERT INTO registry_manifests (tenant, name, version, digest, yaml, key_id, signature)
140                     VALUES ($1, $2, $3, $4, $5, $6, $7)",
141                    &[
142                        &tenant,
143                        &name,
144                        &version,
145                        &digest.to_hex(),
146                        &yaml,
147                        &key_id,
148                        &signature_hex,
149                    ],
150                )
151                .await
152                .map_err(|e| be(&e))?;
153            }
154            // Only the signature columns move; the content is identical by
155            // construction, which is what `decide_publish` established.
156            PublishVerdict::AdoptSignature => {
157                tx.execute(
158                    "UPDATE registry_manifests SET key_id = $4, signature = $5
159                     WHERE tenant = $1 AND name = $2 AND version = $3",
160                    &[&tenant, &name, &version, &key_id, &signature_hex],
161                )
162                .await
163                .map_err(|e| be(&e))?;
164            }
165            PublishVerdict::Unchanged => {}
166        }
167        tx.commit().await.map_err(|e| be(&e))?;
168        Ok(digest)
169    }
170
171    /// One row, or [`RegistryError::NotFound`].
172    async fn row(
173        &self,
174        name: &str,
175        version: &str,
176    ) -> Result<(String, Option<KeySignature>), RegistryError> {
177        let client = self.pool_ref().get().await.map_err(|e| pool_err(&e))?;
178        let row = client
179            .query_opt(
180                "SELECT yaml, key_id, signature FROM registry_manifests
181                 WHERE tenant = $1 AND name = $2 AND version = $3",
182                &[&self.tenant_name(), &name, &version],
183            )
184            .await
185            .map_err(|e| be(&e))?
186            .ok_or_else(|| RegistryError::NotFound {
187                name: name.to_owned(),
188                version: version.to_owned(),
189            })?;
190        Ok((
191            row.get(0),
192            stored_signature(&row.get::<_, String>(1), &row.get::<_, String>(2)),
193        ))
194    }
195}
196
197#[async_trait]
198impl Registry for PostgresStore {
199    async fn publish(&self, manifest: &Manifest) -> Result<Digest, RegistryError> {
200        self.publish_row(manifest, None).await
201    }
202
203    async fn publish_signed(
204        &self,
205        manifest: &Manifest,
206        signer: &dyn Signer,
207    ) -> Result<Digest, RegistryError> {
208        let (_, signature) = sign_manifest(manifest, signer)?;
209        self.publish_row(manifest, Some(signature)).await
210    }
211
212    async fn resolve(&self, name: &str, version: &str) -> Result<Manifest, RegistryError> {
213        let (yaml, _) = self.row(name, version).await?;
214        reparse(name, version, &yaml)
215    }
216
217    async fn resolve_verified(
218        &self,
219        name: &str,
220        version: &str,
221        verifier: &dyn Verifier,
222    ) -> Result<(Manifest, KeyId), RegistryError> {
223        let (yaml, signature) = self.row(name, version).await?;
224        let manifest = reparse(name, version, &yaml)?;
225        let key = check_signature(name, version, &manifest, signature.as_ref(), verifier)?;
226        Ok((manifest, key))
227    }
228
229    async fn versions(&self, name: &str) -> Result<Vec<String>, RegistryError> {
230        let client = self.pool_ref().get().await.map_err(|e| pool_err(&e))?;
231        let rows = client
232            .query(
233                "SELECT version FROM registry_manifests
234                 WHERE tenant = $1 AND name = $2
235                 ORDER BY version",
236                &[&self.tenant_name(), &name],
237            )
238            .await
239            .map_err(|e| be(&e))?;
240        Ok(rows.into_iter().map(|row| row.get(0)).collect())
241    }
242
243    async fn names(&self) -> Result<Vec<String>, RegistryError> {
244        let client = self.pool_ref().get().await.map_err(|e| pool_err(&e))?;
245        let rows = client
246            .query(
247                "SELECT DISTINCT name FROM registry_manifests
248                 WHERE tenant = $1
249                 ORDER BY name",
250                &[&self.tenant_name()],
251            )
252            .await
253            .map_err(|e| be(&e))?;
254        Ok(rows.into_iter().map(|row| row.get(0)).collect())
255    }
256}
257
258/// Not a `StoreError`, and the conversion is here so a caller that has one can
259/// still say what it means.
260impl From<StoreError> for RegistryError {
261    fn from(e: StoreError) -> Self {
262        Self::Backend(e.to_string())
263    }
264}
agentplane/store/postgres_registry.rs

agentplane/store/
postgres_registry.rs
