10 KiB
Utilisation de ksp-store-postgres-lib
1. Quand dépendre directement du backend
Le consumer applicatif normal utilise ksp-store-lib.
Une dépendance directe à ksp-store-postgres-lib est réservée aux composants qui implémentent, intègrent ou testent le bridge physique PostgreSQL. Cette crate ne doit pas devenir une façade Store parallèle.
[dependencies]
ksp-store-api = { path = "../ksp-store-api" }
ksp-store-postgres-lib = { path = "../ksp-store-postgres-lib" }
Le backend reçoit les modèles et traits backend-neutral de ksp-store-api; il ne réexporte pas tokio-postgres, Deadpool ou Rustls.
2. Construire les settings physiques
Pour distinguer création initiale et mise à jour du schéma, utiliser PostgresBackendSettings::with_schema_policy :
fn backend_settings(
network: ksp_store_api::RawNetworkId,
connection_uri: std::string::String,
) -> ksp_store_postgres_lib::PostgresBackendSettings {
return ksp_store_postgres_lib::PostgresBackendSettings::with_schema_policy(
network,
connection_uri,
8,
std::time::Duration::from_secs(10),
std::time::Duration::from_secs(5),
std::time::Duration::from_secs(10),
std::time::Duration::from_secs(5),
ksp_store_postgres_lib::PostgresBackendTlsMode::VerifyFull,
true,
true,
std::time::Duration::from_secs(30),
std::time::Duration::from_secs(10),
);
}
schema_autocreate autorise l'initialisation d'un Store vierge. schema_autoupdate autorise les migrations pending et les réparations additives sûres d'une migration déjà enregistrée. Le constructeur new(..., auto_migrate, ...) existe pour les callers utilisant encore un switch unique et applique cette valeur aux deux politiques.
L'URI est sensible : elle n'est jamais rendue par Debug.
3. Ouvrir, sonder et fermer le backend
async fn use_backend(
settings: ksp_store_postgres_lib::PostgresBackendSettings,
) -> std::result::Result<(), ksp_store_postgres_lib::PostgresBackendError> {
let backend = ksp_store_postgres_lib::PostgresBackend::open(settings).await;
let backend = match backend {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let runtime = backend.runtime_snapshot();
let _capacity = runtime.pool_capacity();
let _size = runtime.pool_size();
let _available = runtime.pool_available();
let _waiting = runtime.pool_waiting();
let health = backend.health().await;
let _ready = health.is_ready();
let _migration_version = health.migration_version();
let _pending = health.pending_migration_count();
let _safe_error_kind = health.error_kind();
return backend.close(std::time::Duration::from_secs(5)).await;
}
open valide la configuration, construit le pool, prouve une connexion et vérifie/applique le bootstrap avant de retourner. close ferme le pool et attend son drain dans la deadline fournie.
Une instance physique est liée à un seul RawNetworkId.
4. Choisir le mode TLS
Pour une connexion PostgreSQL protégée :
ksp_store_postgres_lib::PostgresBackendTlsMode::VerifyFull
VerifyFull impose TLS, les roots système et la vérification de l'identité serveur. Une configuration ne fournissant pas d'identité vérifiable est rejetée.
Pour une topologie explicitement non chiffrée :
ksp_store_postgres_lib::PostgresBackendTlsMode::Disabled
La policy typée choisie par KSP prime sur les paramètres SSL de l'URI.
5. Lire une transaction et ses métadonnées
Les méthodes backend retournent uniquement des modèles ksp-store-api.
async fn read_transaction_state(
backend: &ksp_store_postgres_lib::PostgresBackend,
reference: &ksp_store_api::RawTransactionReference,
) -> std::result::Result<std::option::Option<ksp_store_api::RawTransaction>, ksp_store_postgres_lib::PostgresBackendError> {
let retention = backend.get_raw_transaction_retention_state(reference).await;
if let std::result::Result::Err(error) = retention {
return std::result::Result::Err(error);
}
let tombstone = backend.get_raw_transaction_tombstone(reference).await;
if let std::result::Result::Err(error) = tombstone {
return std::result::Result::Err(error);
}
return backend.get_raw_transaction(reference).await;
}
Full lit le payload chaud, Archived reconstruit le payload depuis l'archive et Purged retourne None. Un tombstone purgé reste lisible séparément.
Un réseau différent de celui du backend est rejeté avant acquisition d'un client du pool.
6. Lire une observation
async fn read_observation(
backend: &ksp_store_postgres_lib::PostgresBackend,
key: &ksp_store_api::RawObservationKey,
) -> std::result::Result<std::option::Option<ksp_store_api::RawTransactionObservation>, ksp_store_postgres_lib::PostgresBackendError> {
return backend.get_raw_transaction_observation(key).await;
}
Les rows PostgreSQL, SQLSTATE, statements et valeurs de bind ne traversent jamais cette API.
7. Persister une acquisition canonique
async fn persist_acquisition(
backend: &ksp_store_postgres_lib::PostgresBackend,
transaction: ksp_store_api::RawTransaction,
observation: ksp_store_api::RawTransactionObservation,
) -> std::result::Result<ksp_store_api::RawAcquisitionWriteOutcome, ksp_store_postgres_lib::PostgresBackendError> {
return backend
.persist_raw_transaction_acquisition(
transaction,
observation,
ksp_store_api::RawTransactionAcquisitionMode::Normal,
)
.await;
}
L'opération est atomique : le canonique et son observation initiale sont tous deux durables ou aucun ne l'est. Une identité déjà présente avec un contenu identique est idempotente ; un contenu divergent retourne PostgresBackendErrorKind::Conflict sans overwrite silencieux.
Pour un tombstone purgé compatible, le mode Normal ne restaure pas le payload. ForceRehydrate doit être demandé explicitement pour rétablir un payload Full.
8. Ajouter une observation à un canonique existant
async fn record_observation(
backend: &ksp_store_postgres_lib::PostgresBackend,
observation: ksp_store_api::RawTransactionObservation,
) -> std::result::Result<ksp_store_api::RawObservationWriteOutcome, ksp_store_postgres_lib::PostgresBackendError> {
return backend.record_raw_transaction_observation(observation).await;
}
Cette opération ne crée jamais la transaction canonique. Une référence absente retourne ReferenceNotFound; une transaction purgée produit l'outcome NotRecorded prévu par l'API.
9. Paginer les transactions
async fn list_transactions(
backend: &ksp_store_postgres_lib::PostgresBackend,
query: &ksp_store_api::RawTransactionQuery,
) -> std::result::Result<ksp_store_api::RawPage<ksp_store_api::RawTransactionReference>, ksp_store_postgres_lib::PostgresBackendError> {
return backend.list_raw_transactions(query).await;
}
La navigation est keyset sur (slot, signature) et exclut les tombstones Purged. Le cursor retourné est opaque et lié au réseau, à la direction et aux bornes de slots de la query qui l'a produit.
Le backend n'utilise pas OFFSET et n'impose pas de plafond métier arbitraire. La seule borne exposée ici provient de la représentation physique de LIMIT + 1 dans PostgreSQL.
10. Appliquer une transition de rétention
async fn apply_retention(
backend: &ksp_store_postgres_lib::PostgresBackend,
transition: ksp_store_api::RawTransactionRetentionTransition,
) -> std::result::Result<ksp_store_api::RawRetentionWriteOutcome, ksp_store_postgres_lib::PostgresBackendError> {
return backend.transition_raw_transaction_retention(transition).await;
}
Le backend applique la transition choisie par le caller ; il ne décide pas de la policy d'éligibilité. Les transitions physiques prises en charge sont Full -> Archived puis Archived -> Purged.
Une transition impliquant Compacted est refusée avec PostgresBackendErrorKind::RetentionCompactionUnsupported tant qu'aucune représentation compactée réelle n'est disponible.
11. Classifier les erreurs sans fuite
fn classify(error: &ksp_store_postgres_lib::PostgresBackendError) {
match error.kind() {
ksp_store_postgres_lib::PostgresBackendErrorKind::ConfigInvalid => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::ConnectFailed => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::Conflict => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::DataInvalid => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::HealthFailed => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::MigrationFailed => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::MigrationMismatch => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::PageLimitUnsupported => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::PoolTimeout => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::QueryInvalid => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::ReadFailed => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::ReferenceNotFound => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::RetentionCompactionUnsupported => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::SchemaNewer => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::ShutdownTimeout => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::TlsFailed => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::WriteFailed => {}
ksp_store_postgres_lib::PostgresBackendErrorKind::WrongNetwork => {}
_ => {}
}
let _safe_phase = error.phase();
}
PostgresBackendError conserve uniquement une classification KSP et une phase statique. Ne pas reconstruire de diagnostic utilisateur à partir d'une erreur brute PostgreSQL.
12. Limites du backend direct
Le backend ne lit aucune variable d'environnement et ne possède aucune sélection de target Config. Les applications, jobs et workers doivent normalement passer par ksp-store-lib.
Les capabilities RawAccount* ne sont pas implémentées par ce backend. Les décisions de batch, priorité, backlog, scheduling et policy de rétention restent hors de sa responsabilité.