Using Rust and Postgres for everything: patterns learned over the years

发表于 更新于 温馨的阅读时光

原文网址Using Rust and Postgres for everything: patterns learned over the years

Intro

作者喜欢简单，可靠，功能又丰富的工具，而在软件领域，就是rust和postgreSQL

随着大模型和编程agent的发展，这两者也变得容易上手，他们兼顾可靠性，功能性，快速，容易适应甚至易维护用他们做的项目，还开源，在作者眼里成了软件工程领域的“end-game”

接下来是尽可能充分的利用rust和postgres的一些开发模式

sqlx

ORMs很糟，用sqlx

sqlx是一个数据库操作依赖，一大核心feature是编译期SQL检查

批处理

在sqlx的0.9版本之后，可以把迭代器绑定作变量使得批量insert或update更容易

原文的example：

impl Repository {
    pub async fn create_users<'c, Q: pg::Queryer<'c>>(&self, db: Q, users: &[User]) -> Result<(), Error> {
        if users.is_empty() {
            return Ok(());
        }

        const QUERY: &str = "INSERT INTO users (id, name)
                SELECT * FROM UNNEST($1::UUID[], $2::TEXT[])";

        sqlx::query(QUERY)
            .bind(users.iter().map(|user| user.id).bind_iter())
            .bind(users.iter().map(|user| user.name).bind_iter())
            .execute(db)
            .await
            .map_err(|err| Error::Unspecified(format!("create_users: {err}")))?;

        return Ok(());
    }

    pub async fn update_users<'c, Q: pg::Queryer<'c>>(&self, db: Q, users: &[User]) -> Result<(), Error> {
        if users.is_empty() {
            return Ok(());
        }

        const QUERY: &str = "UPDATE users
            SET name = user.name
            FROM UNNEST($1::UUID[], $2::TEXT[]) AS user(id, name)
            WHERE users.id = user.id";

        sqlx::query(QUERY)
            .bind(users.iter().map(|user| user.id).bind_iter())
            .bind(users.iter().map(|user| user.name).bind_iter())
            .execute(db)
            .await
            .map_err(|err| Error::Unspecified(format!("update_users: {err}")))?;

        return Ok(());
    }
}

作者贴出代码就不解释了，看来是把工作留给了我

db的类型<'c, Q: pg::Queryer<'c>>是一个实现了查询特征的泛型，不是这里的重点，users是一个User列表切片

insert的时候，UNNEST(array1,array2)是pg特有的写法，会把多个结果展开为一个结果集（粗略看做一张临时表），然后SELECT *，再批量插入

bind的时候，提取user的一个字段，生成一个迭代器，再收集成列表，bind_iter似乎是一个自定义方法

基于PostgreSQL的advisory locks的leader选择

术语很多，其实是个很直观的问题——有些任务只能用一个实例做，如何不出乱子

后端服务通常都被设计成无状态的、可拓展的，如果需要更高的处理能力只需要多开一个实例，但是很多任务只能由一个唯一的实例做，比如调度工作，这就需要一个抢“leader”的机制

作者的代码示例：

/// try_to_become_leader returns a database connection if it successfully becomes leader. The connection
/// MUST BE closed (dropped) to allow other instances to become leader.
/// It blocks until it becomes leader leader or an error happens.
pub async fn try_to_become_leader(
    pool: &Pool,
    advisory_lock_id: i64,
    interval: Duration,
) -> Result<PoolConnection<Postgres>, errs::Error> {
    loop {
        let mut conn = pool.acquire().await.map_err(|err| {
            errs::Error::Internal(format!("error acquiring a Postgres connection from the pool: {err}"))
        })?;

        let is_leader: bool = sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")
            .bind(advisory_lock_id)
            .fetch_one(conn.as_mut())
            .await
            .map_err(|err| errs::Error::Internal(format!("error trying to acquire advisory lock: {err}")))?;
        if is_leader {
            return Ok(conn);
        }

        sleep(interval).await;
    }
}

// 用起来：

pub const SCHEDULER_LEADER_LOCK_ID: i64 = 0;

let _leader_conn = tokio::select! {
    _ = shutdown.changed() => {
        info!("scheduler: Stopping");
        return Ok(());
    }
    leader_conn = pg::try_to_become_leader(&db, pg::SCHEDULER_LEADER_LOCK_ID, Duration::from_secs(1)) => {
        leader_conn?
    },
};

info!("scheduler: is leader");

// run scheduler

作者依旧贴完代码就不解释细节了……

核心是这句:sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")。这里的SELECT是进行一个内部函数调用，名为pg_try_advisory_lock，编号自己指定，其结果是true/false，query_scalar函数返回得到的那一个“值”，或者说将这一个bool值从数据库里提取出来

结果就是只有一个实例可能抢到这个advisory lock，然后由它成为leader，继续接下来的流程

用unlogged tables替换redis

redis用起来有其问题：多了一个有状态的服务需要管理，同时真正需要多少RAM难以估计

所以作者用unlogged tables做快速插入低价值/临时数据

WAL即write ahead log，预写日志，PostgreSQL为了保证数据的安全性，会在实际执行存储前先记录这个动作的日志，以便崩溃后恢复，unlogged table就是不写入WAL的表，因而各种操作更快，更详细的信息引用了另一篇文章

用PostgreSQL作任务队列

用UUID v7 作为索引避免冲突，用FOR UPDATE SKIP LOCKED去pull一个任务

作者给的示例如下，push的时候正常insert,pull的时候筛选出符合条件的任务，然后FOR UPDATE给它们上锁，SKIP LOCKED表示已经锁上了就跳过了

CREATE TABLE queue (
  id UUID PRIMARY KEY,
  created_at TIMESTAMP WITH TIME ZONE NOT NULL,
  updated_at TIMESTAMP WITH TIME ZONE NOT NULL,

  scheduled_for TIMESTAMP WITH TIME ZONE NOT NULL,
  failed_attempts INT NOT NULL,
  status INT NOT NULL,
  message JSONB NOT NULL
);
CREATE INDEX index_queue_on_scheduled_for ON queue (scheduled_for);
CREATE INDEX index_queue_on_status ON queue (status);
#[derive(sqlx::FromRow, Debug, Clone)]
struct Job {
    id: Uuid,
    created_at: DateTime<Utc>,
    updated_at: DateTime<Utc>,

    scheduled_for: DateTime<Utc>,
    failed_attempts: i32,
    status: JobStatus,
    message: Json<JobPayload>,
}

// We use a INT as Postgres representation for performance reasons
#[derive(Debug, Clone, sqlx::Type, PartialEq)]
#[repr(i32)]
enum JobStatus {
    Queued,
    Running,
    Failed,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum JobPayload {
    SendSignInEmail {
        email: String,
        name: String,
        code: String,
    },
    DeleteOldUserData,
    SendNewsletterMessage {
        message_id: Uuid,
    },
    // ...
}


impl Queue {
    async fn push(
        &self,
        job: JobPayload,
        date: Option<DateTime<Utc>>,
    ) -> Result<(), crate::Error> {
        const QUERY: &str = "INSERT INTO queue
            (id, created_at, updated_at, scheduled_for, failed_attempts, status, payload)
            VALUES ($1, $2, $3, $4, $5, $6, $7)";

        let scheduled_for = date.unwrap_or(Utc::now());
        let failed_attempts: i32 = 0;
        let payload = Json(job);
        let status = PostgresJobStatus::Queued;
        let now = Utc::now();
        let job_id = Uuid::new_v7();

        sqlx::query(QUERY)
            .bind(job_id)
            .bind(now)
            .bind(now)
            .bind(scheduled_for)
            .bind(failed_attempts)
            .bind(status)
            .bind(payload)
            .execute(&self.db)
            .await?;
        Ok(())
    }

    async fn pull(&self, number_of_jobs: u32) -> Result<Vec<Job>, crate::Error> {
        const QUERY: &str = "UPDATE queue
            SET status = $1, updated_at = $2
            WHERE id IN (
                SELECT id
                FROM queue
                WHERE status = $3 AND scheduled_for <= $4 AND failed_attempts < $5
                ORDER BY scheduled_for
                FOR UPDATE SKIP LOCKED
                LIMIT $6
            )
            RETURNING *";

        let now = Utc::now();

        let jobs: Vec<PostgresJob> = sqlx::query_as::<_, PostgresJob>(QUERY)
            .bind(PostgresJobStatus::Running)
            .bind(now)
            .bind(PostgresJobStatus::Queued)
            .bind(now)
            .bind(MAX_FAILED_ATTEMPTS)
            .bind(number_of_jobs)
            .fetch_all(&self.db)
            .await?;
        Ok(jobs.into_iter().map(Into::into).collect())
    }

    async fn delete_job(&self, job_id: Uuid) -> Result<(), crate::Error> {
        const QUERY: &str = "DELETE FROM queue WHERE id = $1";

        sqlx::query(QUERY).bind(job_id).execute(&self.db).await?;
        Ok(())
    }

    async fn fail_job(&self, job_id: Uuid) -> Result<(), crate::Error> {
        const QUERY: &str = "UPDATE queue
            SET status = $1, updated_at = $2, failed_attempts = failed_attempts + 1
            WHERE id = $3";

        sqlx::query(QUERY)
            .bind(PostgresJobStatus::Queued)
            .bind(now)
            .bind(job_id)
            .execute(&self.db)
            .await?;
        Ok(())
    }

    async fn clear(&self) -> Result<(), crate::Error> {
        const QUERY: &str = "DELETE FROM queue";

        sqlx::query(query).execute(&self.db).await?;
        Ok(())
    }
}

其它

一个需要认识到的点是，以上介绍的数据库操作的主要优点是方便，降低系统复杂度，有着特定的应用场景，一次数据库操作的开销势必会超过一次内存操作开销几个数量级

# rust# postgreSQL