use sqlx::PgPool;
use std::time::Duration;
use tokio::sync::mpsc;
#[derive(Debug, Clone)]
pub struct MetricPoint {
pub name: String,
pub value: f64,
pub recorded_at: chrono::DateTime<chrono::Utc>,
}
#[derive(Clone)]
pub struct BatchWriter {
tx: mpsc::Sender<MetricPoint>,
}
impl BatchWriter {
pub fn spawn(pool: PgPool, max_batch: usize, flush_interval: Duration) -> Self {
let (tx, rx) = mpsc::channel(max_batch * 4);
tokio::spawn(crate::writer_task::run(pool, rx, max_batch, flush_interval));
BatchWriter { tx }
}
pub async fn submit(&self, point: MetricPoint) -> Result<(), &'static str> {
// Awaits when the channel is full, applying backpressure to producers.
self.tx.send(point).await.map_err(|_| "writer task stopped")
}
}
use crate::metrics::MetricPoint;
use sqlx::{PgPool, QueryBuilder};
use std::time::Duration;
use tokio::sync::mpsc::Receiver;
use tokio::time::{interval, MissedTickBehavior};
pub async fn run(
pool: PgPool,
mut rx: Receiver<MetricPoint>,
max_batch: usize,
flush_interval: Duration,
) {
let mut buf: Vec<MetricPoint> = Vec::with_capacity(max_batch);
let mut ticker = interval(flush_interval);
ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
loop {
tokio::select! {
maybe_point = rx.recv() => {
match maybe_point {
Some(point) => {
buf.push(point);
if buf.len() >= max_batch {
flush_batch(&pool, &mut buf).await;
}
}
None => {
flush_batch(&pool, &mut buf).await;
break;
}
}
}
_ = ticker.tick() => {
if !buf.is_empty() {
flush_batch(&pool, &mut buf).await;
}
}
}
}
}
async fn flush_batch(pool: &PgPool, buf: &mut Vec<MetricPoint>) {
if buf.is_empty() {
return;
}
let mut qb = QueryBuilder::new("INSERT INTO metric_points (name, value, recorded_at) ");
qb.push_values(buf.iter(), |mut b, p| {
b.push_bind(&p.name)
.push_bind(p.value)
.push_bind(p.recorded_at);
});
match qb.build().execute(pool).await {
Ok(res) => tracing::debug!(rows = res.rows_affected(), "flushed batch"),
Err(e) => tracing::error!(error = %e, dropped = buf.len(), "batch flush failed"),
}
buf.clear();
}
CREATE TABLE metric_points (
id BIGGENERATED ALWAYS AS IDENTITY PRIMARY KEY,
name TEXT NOT NULL,
value DOUBLE PRECISION NOT NULL,
recorded_at TIMESTAMPTZ NOT NULL
);
-- BRIN suits append-heavy, time-ordered inserts far better than a B-tree here.
CREATE INDEX metric_points_recorded_at_brin
ON metric_points USING BRIN (recorded_at);
CREATE INDEX metric_points_name_idx
ON metric_points (name);
This snippet shows a common throughput optimization: instead of issuing one INSERT per event, incoming rows are accumulated in memory and flushed to the database in batches, triggered either by a size threshold or by a periodic timer. Writing rows one at a time wastes round-trips and transaction overhead; a multi-row INSERT amortizes that cost dramatically while keeping latency bounded by the flush interval.
In metrics.rs, the MetricPoint struct is the unit of work, and BatchWriter owns the plumbing. Producers never touch the database directly — they call submit, which pushes onto a bounded tokio::mpsc channel. The channel is bounded on purpose: if the writer falls behind, submit awaits capacity, giving the system natural backpressure instead of an unbounded memory blow-up. The BatchWriter::spawn constructor returns a cheap Clone-able handle while the actual draining logic runs on its own task.
The heart of the pattern lives in run, inside writer_task.rs. It uses tokio::select! to race two events: a new message arriving on the receiver, or the flush_interval tick firing. When buf.len() reaches max_batch, it flushes immediately so a burst never grows unbounded; otherwise the interval guarantees that even a single straggling row is persisted within a bounded delay. When the channel closes and recv returns None, the loop performs one final flush and exits cleanly, which matters for graceful shutdown so no buffered data is lost.
flush_batch builds a single parameterized multi-row statement with QueryBuilder::push_values, so all buffered points go in one round-trip. On error it logs and drops the batch rather than blocking the pipeline forever — a deliberate trade-off favoring liveness over strict durability. schema.sql shows the target table with a BRIN index, which suits append-heavy time-series inserts.
The key trade-offs: batching increases throughput but adds up to flush_interval of latency, and a crash can lose the in-memory buffer, so this fits telemetry and metrics rather than financial records. The bounded channel plus size-or-interval flush is the reusable core a developer reaches for whenever high-volume writes dominate.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
struct Config<'a> {
name: &'a str,
value: &'a str,
}
fn parse_config(line: &str) -> Config {Lifetime annotations for flexible borrowing in structs
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
Share this code
Here's the card — post it anywhere.
