Build a Process Logger API in Rust

Introduction

30 Second Summary

Ever wonder how your computer decides which app is hogging all the memory or burning through your CPU? Behind every task manager or monitoring dashboard is a system that watches running processes and keeps track of what they are doing over time.

In this project, you will build a real-time process monitoring API in Rust that polls your Linux system, stores resource usage snapshots in a database, and streams live alerts when processes cross resource thresholds.

What You'll Build

A production-quality backend service that reads live process data from the Linux /proc filesystem, stores it in PostgreSQL, and exposes both a REST API and a WebSocket endpoint for querying and streaming alerts.

By the end of this project, you'll have:

• A background polling daemon that captures CPU and memory usage for every running process every 5 seconds.
• A REST API with cursor-based pagination to query historical process data by time range and process name.
• A WebSocket alert stream that pushes real-time notifications when a process crosses a configurable CPU or memory threshold.
• Secret Mission: Add a remote process termination endpoint that lets you kill runaway processes through the API.

Are there any prerequisites?

This project assumes you have basic Rust knowledge (ownership, traits, async/await) and some familiarity with Axum and SQLx. You also need a running PostgreSQL instance and a Linux system (the /proc filesystem is Linux-only).

Set Up the Project and Dependencies

Before you can poll processes or serve API requests, you need a properly configured Rust project with all the right crates wired up.

In this step, you will initialize a Cargo project and declare every dependency your process logger needs. This includes Axum for HTTP and WebSocket handling, SQLx for PostgreSQL access, and the procfs crate for reading system process information.

In this step, get ready to:

• Initialize a new Cargo binary project.
• Configure all required dependencies in Cargo.toml.
• Verify that every dependency compiles successfully.

Initialize the Cargo project

Cargo is Rust's build system and package manager. Running cargo init inside your existing project directory generates the scaffolding files Rust needs to compile your code.

• Navigate to your project directory by running this command:

cd ~/Desktop/process-logger

• Initialize a new Cargo binary project by running:

cargo init

What does cargo init do?

This command creates a Cargo.toml manifest file and a src/main.rs file with a Hello World program. The Cargo.toml is where you declare your project metadata and dependencies.

• Confirm the project files were created by running:

ls

You should see Cargo.toml, src, and your existing .env file in the output.

Seeing 'already a cargo project'?

If you see this message, it means Cargo detected an existing Cargo.toml in the directory. Delete the file and run cargo init again.

help me troubleshoot cargo init errors

Configure all dependencies in Cargo.toml

Your process logger relies on 17 crates that handle everything from HTTP routing to password hashing. Declaring them all now means you only wait for one compilation cycle.

• Open Cargo.toml in your editor and replace its entire contents with the following:

[package]
name = "process-logger"
version = "0.1.0"
edition = "2021"

[dependencies]
# Web framework with WebSocket support
axum = { version = "0.8", features = ["ws"] }
# Async runtime
tokio = { version = "1", features = ["full"] }
# PostgreSQL driver with compile-time query checking
sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "chrono", "uuid"] }
# Linux /proc filesystem reader
procfs = "0.18"
# Serialization framework
serde = { version = "1", features = ["derive"] }
serde_json = "1"
# JWT token creation and validation
jsonwebtoken = "9"
# Date/time types with serde support
chrono = { version = "0.4", features = ["serde"] }
# UUID generation and serialization
uuid = { version = "1", features = ["v4", "serde"] }
# HTTP middleware stack
tower-http = { version = "0.6", features = ["cors", "trace", "timeout", "compression-gzip"] }
tower = "0.5"
# Structured logging
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
# WebSocket stream utilities
futures-util = "0.3"
# .env file loading
dotenvy = "0.15"
# Password hashing
argon2 = "0.5"
rand_core = { version = "0.6", features = ["std"] }

What are these crates for?

• axum is your web framework. The ws feature enables WebSocket endpoints for real-time alert streaming.
• sqlx connects to PostgreSQL asynchronously. The extra features enable timestamp and UUID column types.
• procfs provides safe, typed access to the Linux /proc filesystem so you can read process CPU and memory data.
• tower-http adds production middleware like CORS, request tracing, timeouts, and gzip compression.
• argon2 handles password hashing with the Argon2id algorithm, the current industry standard.

• Save Cargo.toml.

Compile the dependencies

Rust compiles all dependencies from source the first time you build. This takes a few minutes but only happens once.

• Compile all dependencies by running:

cargo build

This will take 2-5 minutes on the first run. You should see a stream of Compiling lines followed by Finished at the end.

Seeing compilation errors?

If you see a could not find crate error, double-check that you pasted the entire Cargo.toml file including the [package] section at the top.

If you see error[E0463]: can't find crate for openssl-sys, your system is missing OpenSSL dev headers. This project uses tls-rustls instead of tls-native-tls specifically to avoid this. Make sure your Cargo.toml matches exactly.

help me fix cargo build errors in my process-logger project

✔️ Awesome, I've got everything!

Great. Double check you have saved your Cargo.toml file and the build finished with no errors.

ⓧ I'd like to double check the full code

Here is the complete Cargo.toml for comparison:

[package]
name = "process-logger"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = { version = "0.8", features = ["ws"] }
tokio = { version = "1", features = ["full"] }
sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "chrono", "uuid"] }
procfs = "0.18"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
jsonwebtoken = "9"
chrono = { version = "0.4", features = ["serde"] }
uuid = { version = "1", features = ["v4", "serde"] }
tower-http = { version = "0.6", features = ["cors", "trace", "timeout", "compression-gzip"] }
tower = "0.5"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
futures-util = "0.3"
dotenvy = "0.15"
argon2 = "0.5"
rand_core = { version = "0.6", features = ["std"] }

Your project skeleton is ready and all 17 crates compiled cleanly. Next up, you will write the database schema and set up migrations so your process snapshots have a proper home in PostgreSQL.

Create the Database Schema and Migrations

Your Cargo project is ready with all dependencies locked in. Before you can store process data, you need a database schema that can handle thousands of snapshots per hour without slowing down.

In this step, you will design the PostgreSQL tables for storing process snapshots and user accounts. You will also create indexes that make time-range queries fast, even when the table holds millions of rows.

In this step, get ready to:

• Write the SQL migration that creates the users and process_snapshots tables.
• Run the migration against your PostgreSQL database and verify the schema.
• Create the database module file for your Rust project.

Write the initial migration SQL

SQLx uses a migrations/ directory to track database schema changes. Each file represents one migration that runs in order.

• Create the migrations directory inside your process-logger project folder by running this command:

mkdir -p migrations

• Confirm the directory exists by running:

ls

You should see migrations listed alongside src and Cargo.toml.

• Create the migration file by running:

touch migrations/001_initial.sql

• Open migrations/001_initial.sql in your editor and paste the following SQL:

CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    username VARCHAR(255) UNIQUE NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE process_snapshots (
    id BIGSERIAL PRIMARY KEY,
    pid INTEGER NOT NULL,
    name VARCHAR(255) NOT NULL,
    cpu_percent REAL NOT NULL,
    memory_bytes BIGINT NOT NULL,
    captured_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_snapshots_captured_at ON process_snapshots (captured_at);
CREATE INDEX idx_snapshots_name ON process_snapshots (name);
CREATE INDEX idx_snapshots_cpu ON process_snapshots (cpu_percent DESC);

What does this SQL do?

The uuid-ossp extension lets PostgreSQL generate UUIDs natively for user IDs.

The users table stores authentication data. Each user gets a unique UUID, a hashed password, and a creation timestamp.

The process_snapshots table stores one row per process per poll. The three indexes speed up the most common queries: time-range lookups, name filtering, and sorting by CPU usage.

Why BIGSERIAL for the snapshot ID?

Polling every 5 seconds generates a lot of rows. A regular SERIAL (32-bit integer) maxes out at about 2 billion rows. BIGSERIAL (64-bit) gives you effectively unlimited growth for time-series data.

• Save migrations/001_initial.sql.

Run the migration and verify the schema

With the SQL written, you need to run it against your PostgreSQL database. You will use the sqlx CLI to apply migrations.

• Install the SQLx CLI by running:

cargo install sqlx-cli --no-default-features --features postgres

What does this command do?

This installs the sqlx command-line tool with only PostgreSQL support enabled. You will use it to run migrations from the terminal.

Installation taking a long time?

The SQLx CLI compiles from source, which can take a few minutes the first time. This is normal for Rust tooling. Wait for it to finish before continuing.

If you see linker errors, make sure you have libssl-dev and pkg-config installed on your system.

Help me debug my sqlx-cli installation error.

• Run the migration against your database by running:

sqlx migrate run

What does this command do?

The sqlx migrate run command reads the DATABASE_URL from your .env file and applies every SQL file in the migrations/ directory that has not already been applied.

You should see output confirming the migration was applied successfully.

• Verify the tables were created by connecting to your database and running:

psql postgres://postgres:postgres@localhost:5432/process_logger -c "\dt"

You should see users, process_snapshots, and _sqlx_migrations in the output. The _sqlx_migrations table is how SQLx tracks which migrations have already run.

Seeing a connection refused error?

Make sure your PostgreSQL service is running. Check with systemctl status postgresql or pg_isready.

If your PostgreSQL uses a different username or password, update the DATABASE_URL in your .env file to match your credentials.

Help me fix my PostgreSQL connection error.

Create the database module file

Your Rust project needs a db.rs module to keep database-related code organized. For now, this file acts as a placeholder. The actual connection pool will be created in main.rs in a later step.

• Create the module file by running:

touch src/db.rs

• Open src/db.rs and add the following comment:

// Re-export for convenience. Database pool is created in main.rs.
// This module can be extended with helper functions as needed.

Why create an empty module?

Rust projects organize code into modules declared in main.rs with mod db;. Creating the file now means the compiler won't error when you wire up the module declarations in a later step.

• Save src/db.rs.
• Confirm both new files exist by running:

ls migrations/ src/

You should see 001_initial.sql inside migrations/ and db.rs alongside main.rs inside src/.

✔️ Awesome, I've got everything!

Great. Double check you have saved both files.

ⓧ I'd like to double check the full code

Here is the complete migration file for comparison:

CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    username VARCHAR(255) UNIQUE NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE process_snapshots (
    id BIGSERIAL PRIMARY KEY,
    pid INTEGER NOT NULL,
    name VARCHAR(255) NOT NULL,
    cpu_percent REAL NOT NULL,
    memory_bytes BIGINT NOT NULL,
    captured_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_snapshots_captured_at ON process_snapshots (captured_at);
CREATE INDEX idx_snapshots_name ON process_snapshots (name);
CREATE INDEX idx_snapshots_cpu ON process_snapshots (cpu_percent DESC);

And the database module file:

// Re-export for convenience. Database pool is created in main.rs.
// This module can be extended with helper functions as needed.

Your database is ready to receive process data. Next up, you will write the Rust code that reads from your Linux /proc filesystem and inserts snapshots into these tables every 5 seconds.

Poll System Processes with procfs

Your database is ready and waiting for data. The next question is: where does that data come from?

On Linux, the /proc filesystem exposes every running process's CPU time, memory usage, and metadata as readable files. In this step, you'll build a background task that reads /proc every 5 seconds and stores per-process snapshots in PostgreSQL.

In this step, get ready to:

• Define data model structs for process snapshots, alerts, and API requests.
• Write a background polling task that reads /proc and inserts data into PostgreSQL.
• Verify process data is accumulating in the database.

Define your data models

Before writing any logic, you need Rust structs that represent your data. These types define what a process snapshot looks like in the database, what an alert looks like on the wire, and what shape API requests and responses take.

• Create a new file called src/models.rs in your project, then paste the following code:

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

// Database model for registered users
#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
pub struct User {
    pub id: Uuid,
    pub username: String,
    pub password_hash: String,
    pub created_at: DateTime<Utc>,
}

// Database model for process snapshots captured by the poller
#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
pub struct ProcessSnapshot {
    pub id: i64,
    pub pid: i32,
    pub name: String,
    pub cpu_percent: f32,
    pub memory_bytes: i64,
    pub captured_at: DateTime<Utc>,
}

// Alert sent over broadcast channel when a process exceeds thresholds
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProcessAlert {
    pub pid: i32,
    pub name: String,
    pub cpu_percent: f32,
    pub memory_bytes: i64,
    pub alert_type: String,
    pub timestamp: DateTime<Utc>,
}

// Request body for user registration
#[derive(Debug, Deserialize)]
pub struct RegisterRequest {
    pub username: String,
    pub password: String,
}

// Request body for user login
#[derive(Debug, Deserialize)]
pub struct LoginRequest {
    pub username: String,
    pub password: String,
}

// Response returned after successful auth
#[derive(Debug, Serialize)]
pub struct AuthResponse {
    pub token: String,
}

// Query parameters for filtering process snapshots
#[derive(Debug, Deserialize)]
pub struct ProcessQuery {
    pub start: Option<DateTime<Utc>>,
    pub end: Option<DateTime<Utc>>,
    pub name: Option<String>,
    pub cursor: Option<i64>,
    pub limit: Option<i64>,
}

// Paginated response wrapper with a cursor for the next page
#[derive(Debug, Serialize)]
pub struct PaginatedResponse<T: Serialize> {
    pub data: Vec<T>,
    pub next_cursor: Option<i64>,
}

// Query parameters for WebSocket alert thresholds
#[derive(Debug, Deserialize)]
pub struct AlertQuery {
    pub cpu_threshold: Option<f32>,
    pub memory_threshold: Option<i64>,
}

What does this code do?

This file defines all the data types your application will use. They fall into three groups:

• User and ProcessSnapshot map directly to database tables. The sqlx::FromRow derive lets SQLx automatically convert query results into these structs.
• ProcessAlert is the message shape sent over a broadcast channel when a process exceeds CPU or memory thresholds.
• The remaining structs (RegisterRequest, ProcessQuery, etc.) define the shape of API requests and responses you'll build in later steps.

• Save src/models.rs.

Write the process polling module

The procfs crate gives you type-safe access to Linux's /proc filesystem. Instead of manually parsing files like /proc/1234/stat, you call methods that return structured data and handle edge cases like processes dying mid-read.

• Create a new file called src/polling.rs and paste the following code:

use std::sync::Arc;
use chrono::Utc;
use procfs::process::all_processes;
use tokio::time::{interval, Duration};
use tracing::{info, warn};

use crate::models::ProcessAlert;
use crate::AppState;

pub async fn poll_processes(state: Arc<AppState>) {
    let mut ticker = interval(Duration::from_secs(5));
    let cpu_threshold: f32 = 80.0;
    let memory_threshold: i64 = 500_000_000; // 500MB

    info!("Starting process polling every 5 seconds");

    loop {
        ticker.tick().await;

        // Enumerate all processes from /proc
        let processes = match all_processes() {
            Ok(procs) => procs,
            Err(e) => {
                warn!("Failed to read /proc: {}", e);
                continue;
            }
        };

        for proc_entry in processes {
            let proc_entry = match proc_entry {
                Ok(p) => p,
                Err(_) => continue,
            };

            // Read CPU time from /proc/<pid>/stat
            let stat = match proc_entry.stat() {
                Ok(s) => s,
                Err(_) => continue,
            };

            // Read memory info from /proc/<pid>/status
            let status = match proc_entry.status() {
                Ok(s) => s,
                Err(_) => continue,
            };

            let pid = stat.pid;
            let name = stat.comm.clone();
            // Convert clock ticks to CPU seconds (simplified metric)
            let cpu_percent = (stat.utime + stat.stime) as f32 / procfs::ticks_per_second() as f32;
            // VmRSS is reported in kB, convert to bytes
            let memory_bytes = status.vmrss.unwrap_or(0) as i64 * 1024;

            // Insert snapshot into PostgreSQL
            let result = sqlx::query(
                "INSERT INTO process_snapshots (pid, name, cpu_percent, memory_bytes) VALUES ($1, $2, $3, $4)"
            )
            .bind(pid)
            .bind(&name)
            .bind(cpu_percent)
            .bind(memory_bytes)
            .execute(&state.db)
            .await;

            if let Err(e) = result {
                warn!("Failed to insert snapshot for {}: {}", name, e);
            }

            // Check if thresholds are exceeded and send alerts
            if cpu_percent > cpu_threshold || memory_bytes > memory_threshold {
                let alert_type = if cpu_percent > cpu_threshold {
                    "high_cpu".to_string()
                } else {
                    "high_memory".to_string()
                };

                let alert = ProcessAlert {
                    pid,
                    name: name.clone(),
                    cpu_percent,
                    memory_bytes,
                    alert_type,
                    timestamp: Utc::now(),
                };

                // Send alert to broadcast channel (receivers are WebSocket clients)
                let _ = state.alert_tx.send(alert);
            }
        }
    }
}

What does this code do?

• tokio::time::interval(Duration::from_secs(5)) creates a repeating timer that fires every 5 seconds. The loop waits on each tick before scanning processes.
• all_processes() reads every entry in /proc. Each entry is a Result because processes can vanish between listing and reading.
• stat.utime + stat.stime gives total CPU clock ticks. Dividing by ticks_per_second() converts to cumulative CPU seconds. This is a simplified metric, not an instantaneous percentage.
• When a process exceeds 80% CPU or 500MB memory, the function sends a ProcessAlert through the broadcast channel. WebSocket clients will subscribe to this channel in a later step.

Why use procfs instead of parsing /proc manually?

You could read and parse /proc/<pid>/stat files yourself, but the format is fragile and full of edge cases. The procfs crate handles processes disappearing mid-read, fields with embedded spaces, and kernel version differences for you.

• Save src/polling.rs.

Run the poller and verify data accumulates

The polling module references an AppState struct that holds the database pool and broadcast channel. You need a main.rs that creates this state and spawns the polling task.

• Open src/main.rs and replace any existing content with the following code:

use std::net::SocketAddr;
use std::sync::Arc;

use sqlx::postgres::PgPoolOptions;
use tokio::sync::broadcast;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};

mod db;
mod models;
mod polling;

// Shared application state passed to handlers and background tasks
pub struct AppState {
    pub db: sqlx::PgPool,
    pub jwt_secret: String,
    pub alert_tx: broadcast::Sender<models::ProcessAlert>,
}

#[tokio::main]
async fn main() {
    dotenvy::dotenv().ok();

    // Initialize structured logging with env-based filtering
    tracing_subscriber::registry()
        .with(EnvFilter::try_from_default_env().unwrap_or_else(|_| {
            "process_logger=debug,tower_http=debug".into()
        }))
        .with(tracing_subscriber::fmt::layer())
        .init();

    let database_url = std::env::var("DATABASE_URL")
        .expect("DATABASE_URL must be set");
    let jwt_secret = std::env::var("JWT_SECRET")
        .unwrap_or_else(|_| "super-secret-key-change-me".to_string());

    // Create database connection pool
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&database_url)
        .await
        .expect("Failed to connect to database");

    // Run pending migrations
    sqlx::migrate!("./migrations")
        .run(&pool)
        .await
        .expect("Failed to run migrations");

    // Broadcast channel for process alerts (capacity 100 messages)
    let (alert_tx, _) = broadcast::channel(100);

    let state = Arc::new(AppState {
        db: pool.clone(),
        jwt_secret,
        alert_tx: alert_tx.clone(),
    });

    // Start background polling task
    let polling_state = state.clone();
    tokio::spawn(async move {
        polling::poll_processes(polling_state).await;
    });

    // Keep the process running so polling continues
    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
    tracing::info!("Server listening on {}", addr);
    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
    axum::serve(listener, axum::Router::new().into_make_service())
        .await
        .unwrap();
}

What does this code do?

• AppState bundles the database pool, JWT secret, and broadcast channel into one shared object. Wrapping it in Arc lets multiple tasks access it concurrently.
• tokio::spawn launches the polling function as a background task that runs independently of the main server.
• The empty axum::Router keeps the process alive while polling runs. You will add API routes to this router in the next step.

• Save src/main.rs.
• Start the application by running this command:

cargo run

You should see log output including Starting process polling every 5 seconds followed by Server listening on 127.0.0.1:3000.

First compile takes a while

Rust compiles all dependencies on the first build. This can take 2-3 minutes depending on your machine. Subsequent builds are much faster since only changed code recompiles.

• Wait about 10-15 seconds so the poller runs 2-3 times, then open a new terminal window.
• Query the database to confirm snapshots are being inserted by running:

psql -d process_logger -c "SELECT count(*) FROM process_snapshots;"

You should see a count greater than zero. If you wait another 5 seconds and run the query again, the count should increase.

• Inspect a few actual rows by running:

psql -d process_logger -c "SELECT pid, name, cpu_percent, memory_bytes, captured_at FROM process_snapshots ORDER BY captured_at DESC LIMIT 5;"

You should see 5 rows with real process names, PIDs, and timestamps from the last few seconds.

Seeing zero rows or permission errors?

• If the count is 0 but the server is running, check that you waited at least 5 seconds. The first tick fires immediately but reading all processes takes a moment.
• If you see Failed to read /proc warnings in the server output, you may be running inside a container that restricts /proc access. Run directly on your Linux host instead.
• If you see Failed to connect to database on startup, double-check your .env file has the correct DATABASE_URL and that PostgreSQL is running.

Help me debug my process polling task

• Stop the running server by pressing Ctrl+C in the terminal where cargo run is running.

✔️ Awesome, I've got everything!

Great. Double check you have saved both src/models.rs and src/polling.rs.

ⓧ I'd like to double check the full code

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
pub struct User {
    pub id: Uuid,
    pub username: String,
    pub password_hash: String,
    pub created_at: DateTime<Utc>,
}

#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
pub struct ProcessSnapshot {
    pub id: i64,
    pub pid: i32,
    pub name: String,
    pub cpu_percent: f32,
    pub memory_bytes: i64,
    pub captured_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProcessAlert {
    pub pid: i32,
    pub name: String,
    pub cpu_percent: f32,
    pub memory_bytes: i64,
    pub alert_type: String,
    pub timestamp: DateTime<Utc>,
}

#[derive(Debug, Deserialize)]
pub struct RegisterRequest {
    pub username: String,
    pub password: String,
}

#[derive(Debug, Deserialize)]
pub struct LoginRequest {
    pub username: String,
    pub password: String,
}

#[derive(Debug, Serialize)]
pub struct AuthResponse {
    pub token: String,
}

#[derive(Debug, Deserialize)]
pub struct ProcessQuery {
    pub start: Option<DateTime<Utc>>,
    pub end: Option<DateTime<Utc>>,
    pub name: Option<String>,
    pub cursor: Option<i64>,
    pub limit: Option<i64>,
}

#[derive(Debug, Serialize)]
pub struct PaginatedResponse<T: Serialize> {
    pub data: Vec<T>,
    pub next_cursor: Option<i64>,
}

#[derive(Debug, Deserialize)]
pub struct AlertQuery {
    pub cpu_threshold: Option<f32>,
    pub memory_threshold: Option<i64>,
}

use std::sync::Arc;
use chrono::Utc;
use procfs::process::all_processes;
use tokio::time::{interval, Duration};
use tracing::{info, warn};

use crate::models::ProcessAlert;
use crate::AppState;

pub async fn poll_processes(state: Arc<AppState>) {
    let mut ticker = interval(Duration::from_secs(5));
    let cpu_threshold: f32 = 80.0;
    let memory_threshold: i64 = 500_000_000;

    info!("Starting process polling every 5 seconds");

    loop {
        ticker.tick().await;

        let processes = match all_processes() {
            Ok(procs) => procs,
            Err(e) => {
                warn!("Failed to read /proc: {}", e);
                continue;
            }
        };

        for proc_entry in processes {
            let proc_entry = match proc_entry {
                Ok(p) => p,
                Err(_) => continue,
            };

            let stat = match proc_entry.stat() {
                Ok(s) => s,
                Err(_) => continue,
            };

            let status = match proc_entry.status() {
                Ok(s) => s,
                Err(_) => continue,
            };

            let pid = stat.pid;
            let name = stat.comm.clone();
            let cpu_percent = (stat.utime + stat.stime) as f32 / procfs::ticks_per_second() as f32;
            let memory_bytes = status.vmrss.unwrap_or(0) as i64 * 1024;

            let result = sqlx::query(
                "INSERT INTO process_snapshots (pid, name, cpu_percent, memory_bytes) VALUES ($1, $2, $3, $4)"
            )
            .bind(pid)
            .bind(&name)
            .bind(cpu_percent)
            .bind(memory_bytes)
            .execute(&state.db)
            .await;

            if let Err(e) = result {
                warn!("Failed to insert snapshot for {}: {}", name, e);
            }

            if cpu_percent > cpu_threshold || memory_bytes > memory_threshold {
                let alert_type = if cpu_percent > cpu_threshold {
                    "high_cpu".to_string()
                } else {
                    "high_memory".to_string()
                };

                let alert = ProcessAlert {
                    pid,
                    name: name.clone(),
                    cpu_percent,
                    memory_bytes,
                    alert_type,
                    timestamp: Utc::now(),
                };

                let _ = state.alert_tx.send(alert);
            }
        }
    }
}

Your polling daemon is capturing live process data. Next up, you'll build a REST API on top of this data so you can query what was running at any point in time.

Build the REST API with Query Filters

Your background poller is steadily filling the process_snapshots table with data. That data is useless without a way to query it.

In this step, you will build the REST API layer that lets clients ask questions like "what was consuming resources between 2am and 4am?" You will also add cursor-based pagination so results come back in manageable pages.

In this step, get ready to:

• Write a GET /api/processes handler with time-range and name filters.
• Implement cursor-based pagination for large result sets.
• Add a GET /api/processes/top endpoint returning the top 10 processes by CPU.

Create the process query handler

The first endpoint needs to support flexible querying. Clients should be able to filter by time range, filter by process name, and page through results without performance degrading on large datasets.

Why cursor-based pagination?

Offset-based pagination (OFFSET 50 LIMIT 10) gets slower as the offset grows, because the database still reads and discards all skipped rows. It also suffers from "page drift" when new rows are inserted between requests.

Cursor-based pagination uses a unique, ordered column (the id column) as a bookmark. Each response returns the last ID seen, and the next request starts from that point. This gives consistent O(1) performance regardless of how deep into the dataset you are.

• Create a new file called src/handlers.rs in your project.
• Add the following imports and the get_processes handler function:

use axum::{
    extract::{Query, State, WebSocketUpgrade},
    extract::ws::{Message, WebSocket},
    http::StatusCode,
    response::{IntoResponse, Response},
    Json,
};
use futures_util::{SinkExt, StreamExt};
use std::sync::Arc;

use crate::auth;
use crate::models::*;
use crate::AppState;

// Handler for GET /api/processes with optional time range, name filter, and cursor pagination
pub async fn get_processes(
    State(state): State<Arc<AppState>>,
    Query(query): Query<ProcessQuery>,
) -> Result<Json<PaginatedResponse<ProcessSnapshot>>, StatusCode> {
    // Cap limit at 100 to prevent huge result sets
    let limit = query.limit.unwrap_or(50).min(100);
    // Start from cursor position (0 means beginning)
    let cursor = query.cursor.unwrap_or(0);

    let snapshots = sqlx::query_as::<_, ProcessSnapshot>(
        "SELECT * FROM process_snapshots
         WHERE id > $1
         AND ($2::timestamptz IS NULL OR captured_at >= $2)
         AND ($3::timestamptz IS NULL OR captured_at <= $3)
         AND ($4::text IS NULL OR name ILIKE $4)
         ORDER BY id ASC
         LIMIT $5"
    )
    .bind(cursor)
    .bind(query.start)
    .bind(query.end)
    .bind(query.name.map(|n| format!("%{}%", n)))
    .bind(limit)
    .fetch_all(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    // Return the last ID as the cursor for the next page
    let next_cursor = snapshots.last().map(|s| s.id);

    Ok(Json(PaginatedResponse {
        data: snapshots,
        next_cursor,
    }))
}

What does this code do?

• The Query(query) extractor deserializes URL query parameters into your ProcessQuery struct automatically.
• The SQL uses IS NULL OR patterns so that filters are only applied when the client actually provides them. Omitting a parameter means "no filter on this field."
• The ILIKE operator does case-insensitive substring matching on process names.
• The next_cursor field gives the client the ID to pass in their next request to fetch the following page.

• Save src/handlers.rs.

You will not be able to compile yet because the router and other modules are not wired up. That comes at the end of this step.

Add the top processes endpoint

The second endpoint gives a quick snapshot of what is currently consuming the most CPU. It uses PostgreSQL's DISTINCT ON to return only the most recent snapshot per unique PID.

• In src/handlers.rs, add the following function below get_processes:

// Handler for GET /api/processes/top - returns the 10 most recent snapshots, one per PID
pub async fn get_top_processes(
    State(state): State<Arc<AppState>>,
) -> Result<Json<Vec<ProcessSnapshot>>, StatusCode> {
    let snapshots = sqlx::query_as::<_, ProcessSnapshot>(
        "SELECT DISTINCT ON (pid) * FROM process_snapshots
         ORDER BY pid, captured_at DESC
         LIMIT 10"
    )
    .fetch_all(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(snapshots))
}

What does this code do?

The DISTINCT ON (pid) clause is a PostgreSQL extension that returns only the first row for each unique PID value. Combined with ORDER BY pid, captured_at DESC, it picks the most recent snapshot for each process.

• Save src/handlers.rs.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before moving on.

ⓧ I'd like to double check the full code

use axum::{
    extract::{Query, State, WebSocketUpgrade},
    extract::ws::{Message, WebSocket},
    http::StatusCode,
    response::{IntoResponse, Response},
    Json,
};
use futures_util::{SinkExt, StreamExt};
use std::sync::Arc;

use crate::auth;
use crate::models::*;
use crate::AppState;

pub async fn get_processes(
    State(state): State<Arc<AppState>>,
    Query(query): Query<ProcessQuery>,
) -> Result<Json<PaginatedResponse<ProcessSnapshot>>, StatusCode> {
    let limit = query.limit.unwrap_or(50).min(100);
    let cursor = query.cursor.unwrap_or(0);

    let snapshots = sqlx::query_as::<_, ProcessSnapshot>(
        "SELECT * FROM process_snapshots
         WHERE id > $1
         AND ($2::timestamptz IS NULL OR captured_at >= $2)
         AND ($3::timestamptz IS NULL OR captured_at <= $3)
         AND ($4::text IS NULL OR name ILIKE $4)
         ORDER BY id ASC
         LIMIT $5"
    )
    .bind(cursor)
    .bind(query.start)
    .bind(query.end)
    .bind(query.name.map(|n| format!("%{}%", n)))
    .bind(limit)
    .fetch_all(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    let next_cursor = snapshots.last().map(|s| s.id);

    Ok(Json(PaginatedResponse {
        data: snapshots,
        next_cursor,
    }))
}

pub async fn get_top_processes(
    State(state): State<Arc<AppState>>,
) -> Result<Json<Vec<ProcessSnapshot>>, StatusCode> {
    let snapshots = sqlx::query_as::<_, ProcessSnapshot>(
        "SELECT DISTINCT ON (pid) * FROM process_snapshots
         ORDER BY pid, captured_at DESC
         LIMIT 10"
    )
    .fetch_all(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(snapshots))
}

Wire the Axum router and start the server

Before the router can compile, you need to create the auth module that handlers.rs imports. You will fully use this module in the next step when you add JWT authentication. For now, it provides the token creation and verification functions your app will rely on.

• Create a new file called src/auth.rs and add the following code:

use chrono::{Duration, Utc};
use jsonwebtoken::{decode, encode, DecodingKey, EncodingKey, Header, Validation};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

// JWT claims payload - stored inside every issued token
#[derive(Debug, Serialize, Deserialize)]
pub struct Claims {
    pub sub: String,       // username
    pub user_id: String,   // UUID as string
    pub exp: usize,        // expiration timestamp
}

// Create a signed JWT token valid for 24 hours
pub fn create_token(user_id: Uuid, username: &str, secret: &str) -> Result<String, jsonwebtoken::errors::Error> {
    let expiration = Utc::now()
        .checked_add_signed(Duration::hours(24))
        .expect("valid timestamp")
        .timestamp() as usize;

    let claims = Claims {
        sub: username.to_string(),
        user_id: user_id.to_string(),
        exp: expiration,
    };

    encode(
        &Header::default(),
        &claims,
        &EncodingKey::from_secret(secret.as_bytes()),
    )
}

// Verify and decode a JWT token, returning the claims if valid
pub fn verify_token(token: &str, secret: &str) -> Result<Claims, jsonwebtoken::errors::Error> {
    let token_data = decode::<Claims>(
        token,
        &DecodingKey::from_secret(secret.as_bytes()),
        &Validation::default(),
    )?;
    Ok(token_data.claims)
}

What does this code do?

• The Claims struct defines what data lives inside every JWT. The exp field tells the verifier when the token expires.
• The create_token function encodes claims into a signed token using HMAC-SHA256 (the default algorithm).
• The verify_token function decodes the token and checks the signature and expiration automatically.

• Save src/auth.rs.

Now you can wire everything together in the main entry point. The router maps URL paths to handler functions and starts the server.

• Open src/main.rs and replace its contents with the following:

use std::net::SocketAddr;
use std::sync::Arc;

use axum::{routing::get, Router};
use sqlx::postgres::PgPoolOptions;
use tokio::sync::broadcast;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};

mod auth;
mod db;
mod handlers;
mod models;
mod polling;

pub struct AppState {
    pub db: sqlx::PgPool,
    pub jwt_secret: String,
    pub alert_tx: broadcast::Sender<models::ProcessAlert>,
}

#[tokio::main]
async fn main() {
    dotenvy::dotenv().ok();

    // Initialize structured logging with env-based filter
    tracing_subscriber::registry()
        .with(EnvFilter::try_from_default_env().unwrap_or_else(|_| {
            "process_logger=debug,tower_http=debug".into()
        }))
        .with(tracing_subscriber::fmt::layer())
        .init();

    let database_url = std::env::var("DATABASE_URL")
        .expect("DATABASE_URL must be set");
    let jwt_secret = std::env::var("JWT_SECRET")
        .unwrap_or_else(|_| "super-secret-key-change-me".to_string());

    // Create a connection pool with max 5 concurrent connections
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&database_url)
        .await
        .expect("Failed to connect to database");

    // Run pending migrations on startup
    sqlx::migrate!("./migrations")
        .run(&pool)
        .await
        .expect("Failed to run migrations");

    // Broadcast channel for real-time alerts (used later for WebSocket)
    let (alert_tx, _) = broadcast::channel(100);

    let state = Arc::new(AppState {
        db: pool.clone(),
        jwt_secret,
        alert_tx: alert_tx.clone(),
    });

    // Start background polling task
    let polling_state = state.clone();
    tokio::spawn(async move {
        polling::poll_processes(polling_state).await;
    });

    // Build the router with process query endpoints
    let app = Router::new()
        .route("/api/processes", get(handlers::get_processes))
        .route("/api/processes/top", get(handlers::get_top_processes))
        .with_state(state);

    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
    tracing::info!("Server listening on {}", addr);
    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

What does this code do?

• The PgPoolOptions::new().max_connections(5) creates a database connection pool. Multiple handlers can query the database concurrently without opening new connections each time.
• The sqlx::migrate! macro embeds your migration SQL at compile time and runs any pending migrations on server startup.
• The broadcast::channel(100) creates a multi-producer, multi-consumer channel with a buffer of 100 messages. You will use this in a later step for WebSocket alerts.
• The tokio::spawn launches the polling task as an independent async task that runs alongside the HTTP server.

• Save src/main.rs.
• Build and run the server by running this command:

cargo run

You should see log output including Server listening on 127.0.0.1:3000 and Starting process polling every 5 seconds.

• Open a second terminal and test the processes endpoint by running:

curl http://localhost:3000/api/processes?limit=5

You should see a JSON response containing a data array of process snapshots and a next_cursor value.

• Test the top processes endpoint by running:

curl http://localhost:3000/api/processes/top

You should see a JSON array of up to 10 process snapshots, each with a unique PID.

• Test pagination by passing the next_cursor value from your first response:

curl "http://localhost:3000/api/processes?limit=5&cursor=5"

You should see the next page of results starting after the cursor ID.

Seeing an empty data array?

• The polling task needs a few seconds to populate the database. Wait 10 seconds after starting the server, then try again.
• Confirm data exists by running psql -d process_logger -c "SELECT count(*) FROM process_snapshots;" in a separate terminal. If the count is 0, check for permission warnings in the server log.
• Make sure your .env file has the correct DATABASE_URL pointing to your process_logger database.

Help me debug why my process logger API returns empty results.

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before continuing.

ⓧ I'd like to double check the full code

use std::net::SocketAddr;
use std::sync::Arc;

use axum::{routing::get, Router};
use sqlx::postgres::PgPoolOptions;
use tokio::sync::broadcast;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};

mod auth;
mod db;
mod handlers;
mod models;
mod polling;

pub struct AppState {
    pub db: sqlx::PgPool,
    pub jwt_secret: String,
    pub alert_tx: broadcast::Sender<models::ProcessAlert>,
}

#[tokio::main]
async fn main() {
    dotenvy::dotenv().ok();

    tracing_subscriber::registry()
        .with(EnvFilter::try_from_default_env().unwrap_or_else(|_| {
            "process_logger=debug,tower_http=debug".into()
        }))
        .with(tracing_subscriber::fmt::layer())
        .init();

    let database_url = std::env::var("DATABASE_URL")
        .expect("DATABASE_URL must be set");
    let jwt_secret = std::env::var("JWT_SECRET")
        .unwrap_or_else(|_| "super-secret-key-change-me".to_string());

    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&database_url)
        .await
        .expect("Failed to connect to database");

    sqlx::migrate!("./migrations")
        .run(&pool)
        .await
        .expect("Failed to run migrations");

    let (alert_tx, _) = broadcast::channel(100);

    let state = Arc::new(AppState {
        db: pool.clone(),
        jwt_secret,
        alert_tx: alert_tx.clone(),
    });

    let polling_state = state.clone();
    tokio::spawn(async move {
        polling::poll_processes(polling_state).await;
    });

    let app = Router::new()
        .route("/api/processes", get(handlers::get_processes))
        .route("/api/processes/top", get(handlers::get_top_processes))
        .with_state(state);

    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
    tracing::info!("Server listening on {}", addr);
    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

use chrono::{Duration, Utc};
use jsonwebtoken::{decode, encode, DecodingKey, EncodingKey, Header, Validation};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

#[derive(Debug, Serialize, Deserialize)]
pub struct Claims {
    pub sub: String,
    pub user_id: String,
    pub exp: usize,
}

pub fn create_token(user_id: Uuid, username: &str, secret: &str) -> Result<String, jsonwebtoken::errors::Error> {
    let expiration = Utc::now()
        .checked_add_signed(Duration::hours(24))
        .expect("valid timestamp")
        .timestamp() as usize;

    let claims = Claims {
        sub: username.to_string(),
        user_id: user_id.to_string(),
        exp: expiration,
    };

    encode(
        &Header::default(),
        &claims,
        &EncodingKey::from_secret(secret.as_bytes()),
    )
}

pub fn verify_token(token: &str, secret: &str) -> Result<Claims, jsonwebtoken::errors::Error> {
    let token_data = decode::<Claims>(
        token,
        &DecodingKey::from_secret(secret.as_bytes()),
        &Validation::default(),
    )?;
    Ok(token_data.claims)
}

use axum::{
    extract::{Query, State, WebSocketUpgrade},
    extract::ws::{Message, WebSocket},
    http::StatusCode,
    response::{IntoResponse, Response},
    Json,
};
use futures_util::{SinkExt, StreamExt};
use std::sync::Arc;

use crate::auth;
use crate::models::*;
use crate::AppState;

pub async fn get_processes(
    State(state): State<Arc<AppState>>,
    Query(query): Query<ProcessQuery>,
) -> Result<Json<PaginatedResponse<ProcessSnapshot>>, StatusCode> {
    let limit = query.limit.unwrap_or(50).min(100);
    let cursor = query.cursor.unwrap_or(0);

    let snapshots = sqlx::query_as::<_, ProcessSnapshot>(
        "SELECT * FROM process_snapshots
         WHERE id > $1
         AND ($2::timestamptz IS NULL OR captured_at >= $2)
         AND ($3::timestamptz IS NULL OR captured_at <= $3)
         AND ($4::text IS NULL OR name ILIKE $4)
         ORDER BY id ASC
         LIMIT $5"
    )
    .bind(cursor)
    .bind(query.start)
    .bind(query.end)
    .bind(query.name.map(|n| format!("%{}%", n)))
    .bind(limit)
    .fetch_all(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    let next_cursor = snapshots.last().map(|s| s.id);

    Ok(Json(PaginatedResponse {
        data: snapshots,
        next_cursor,
    }))
}

pub async fn get_top_processes(
    State(state): State<Arc<AppState>>,
) -> Result<Json<Vec<ProcessSnapshot>>, StatusCode> {
    let snapshots = sqlx::query_as::<_, ProcessSnapshot>(
        "SELECT DISTINCT ON (pid) * FROM process_snapshots
         ORDER BY pid, captured_at DESC
         LIMIT 10"
    )
    .fetch_all(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(snapshots))
}

Your API is serving data, but anyone on the network can hit these endpoints without authenticating. Next up, you will lock everything down with JWT tokens so only registered users can query process data.

Add JWT Authentication Middleware

Your API currently lets anyone query process data without identifying themselves. In a production monitoring system, you need to know who is accessing sensitive system information.

In this step, you will add user registration, login, and a JWT authentication layer. Clients will send a token with every request to prove their identity.

In this step, get ready to:

• Create register and login handlers that issue JWT tokens.
• Write a Tower middleware layer that validates JWTs on protected routes.
• Test the full authentication flow with curl.

Create the register handler

The register handler accepts a username and password, hashes the password with Argon2 (an industry-standard algorithm), stores the user in PostgreSQL, and returns a signed JWT.

• In src/handlers.rs, find the existing imports at the top of the file. Replace them with this updated import block that adds the auth module reference:

use axum::{
    extract::{Query, State, WebSocketUpgrade},
    extract::ws::{Message, WebSocket},
    http::StatusCode,
    response::{IntoResponse, Response},
    Json,
};
use futures_util::{SinkExt, StreamExt};
use std::sync::Arc;

use crate::auth;
use crate::models::*;
use crate::AppState;

What's happening here?

The imports are the same as before. You already have these in your file from the previous step. The key import for authentication is the crate::auth module you wrote in Step 2, which provides create_token and verify_token.

• Below the existing use statements and above the get_processes function, add the register handler:

pub async fn register(
    State(state): State<Arc<AppState>>,
    Json(body): Json<RegisterRequest>,
) -> Result<Json<AuthResponse>, StatusCode> {
    use argon2::{Argon2, PasswordHasher};
    use argon2::password_hash::SaltString;
    use rand_core::OsRng;

    // Generate a random salt and hash the password with Argon2
    let salt = SaltString::generate(&mut OsRng);
    let argon2 = Argon2::default();
    let password_hash = argon2
        .hash_password(body.password.as_bytes(), &salt)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
        .to_string();

    // Insert the new user and return the full row
    let user = sqlx::query_as::<_, User>(
        "INSERT INTO users (username, password_hash) VALUES ($1, $2) RETURNING *"
    )
    .bind(&body.username)
    .bind(&password_hash)
    .fetch_one(&state.db)
    .await
    .map_err(|_| StatusCode::CONFLICT)?;

    // Issue a JWT for the newly registered user
    let token = auth::create_token(user.id, &user.username, &state.jwt_secret)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(AuthResponse { token }))
}

What does this code do?

• SaltString::generate(&mut OsRng) creates a cryptographically random salt for each user. This ensures identical passwords produce different hashes.
• Argon2::default() uses the Argon2id variant, which is resistant to both side-channel and GPU attacks.
• The RETURNING * clause gives back the full user row (including the generated UUID) without a second query.
• If the username already exists, PostgreSQL raises a unique constraint violation, which maps to StatusCode::CONFLICT (HTTP 409).

• Save src/handlers.rs.

You won't see a change yet. The login handler comes next, and then you can test both together.

Create the login handler

The login handler looks up the user by username, verifies the password against the stored hash, and issues a fresh JWT on success.

• In src/handlers.rs, add the login handler directly below the register function you just wrote:

pub async fn login(
    State(state): State<Arc<AppState>>,
    Json(body): Json<LoginRequest>,
) -> Result<Json<AuthResponse>, StatusCode> {
    use argon2::{Argon2, PasswordVerifier, PasswordHash};

    // Look up the user by username
    let user = sqlx::query_as::<_, User>(
        "SELECT * FROM users WHERE username = $1"
    )
    .bind(&body.username)
    .fetch_optional(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
    .ok_or(StatusCode::UNAUTHORIZED)?;

    // Verify the provided password against the stored hash
    let parsed_hash = PasswordHash::new(&user.password_hash)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Argon2::default()
        .verify_password(body.password.as_bytes(), &parsed_hash)
        .map_err(|_| StatusCode::UNAUTHORIZED)?;

    // Issue a JWT for the authenticated user
    let token = auth::create_token(user.id, &user.username, &state.jwt_secret)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(AuthResponse { token }))
}

What does this code do?

• fetch_optional returns None if the username does not exist. The .ok_or(StatusCode::UNAUTHORIZED) converts that into a 401 response.
• PasswordHash::new parses the stored PHC-format string back into a structured hash object that Argon2 can verify against.
• verify_password compares the raw password bytes to the stored hash. If they don't match, the request gets a 401.

Why Argon2 instead of bcrypt?

Argon2id (the default variant in the argon2 crate) is the current recommendation for password hashing. It resists both GPU-based brute force and side-channel timing attacks. For new Rust projects, it is a better choice than bcrypt.

• Save src/handlers.rs.

The register and login routes are already declared in your router from the previous step. Next you will write the middleware that validates tokens on protected routes.

Write the auth middleware

A Tower middleware intercepts every request before it reaches your handler. Your auth middleware will extract the Authorization: Bearer <token> header, verify the JWT signature, and reject the request with a 401 if the token is invalid or missing.

• Open src/middleware.rs (this file was declared in main.rs but is currently empty) and paste the following:

use axum::{
    extract::Request,
    http::StatusCode,
    middleware::Next,
    response::Response,
};
use std::sync::Arc;

use crate::auth;
use crate::AppState;

pub async fn auth_middleware(
    state: Arc<AppState>,
    request: Request,
    next: Next,
) -> Result<Response, StatusCode> {
    // Extract the Authorization header value
    let auth_header = request
        .headers()
        .get("Authorization")
        .and_then(|value| value.to_str().ok());

    // Strip the "Bearer " prefix to get the raw token
    let token = match auth_header {
        Some(header) if header.starts_with("Bearer ") => &header[7..],
        _ => return Err(StatusCode::UNAUTHORIZED),
    };

    // Verify the token signature and expiry
    match auth::verify_token(token, &state.jwt_secret) {
        Ok(_claims) => Ok(next.run(request).await),
        Err(_) => Err(StatusCode::UNAUTHORIZED),
    }
}

What does this code do?

• The function signature (state, request, next) follows the pattern Axum expects for from_fn_with_state middleware. Axum injects the shared state automatically.
• &header[7..] slices off the first 7 characters ("Bearer ") to isolate the raw JWT string.
• If verification succeeds, next.run(request).await passes the request through to the actual handler. If it fails, the request never reaches the handler.

• Save src/middleware.rs.

Applying the middleware to specific routes

You can protect specific route groups by calling .route_layer(axum::middleware::from_fn_with_state(state.clone(), middleware::auth_middleware)) on a nested Router. The middleware is ready to use whenever you need route-level protection.

Seeing a compile error about middleware?

Make sure the function signature uses state: Arc<AppState> (not State(state) extractor syntax). Middleware functions receive the state directly from from_fn_with_state, not through Axum's extractor pattern.

help me fix a compile error in my auth middleware

Test the authentication flow

Time to verify the full register-login-token cycle works end to end.

• Compile and run the server by running this command in your terminal:

cargo run

• You should see Server listening on 127.0.0.1:3000 in the terminal output.
• Open a second terminal window. Register a new user by running:

curl -X POST http://127.0.0.1:3000/api/register \
  -H "Content-Type: application/json" \
  -d '{"username": "testuser", "password": "testpass123"}'

• You should see a JSON response containing a token field with a long encoded string. That is your JWT.
• Test logging in with the same credentials by running:

curl -X POST http://127.0.0.1:3000/api/login \
  -H "Content-Type: application/json" \
  -d '{"username": "testuser", "password": "testpass123"}'

• You should see another JSON response with a fresh token value.
• Verify that a wrong password returns an error by running:

curl -X POST http://127.0.0.1:3000/api/login \
  -H "Content-Type: application/json" \
  -d '{"username": "testuser", "password": "wrongpassword"}'

• You should see an empty response with HTTP status 401 (Unauthorized). This confirms password verification is working.

Getting a 500 Internal Server Error on register?

Check that your PostgreSQL database is running and that the users table exists. Run psql process_logger -c '\dt' to confirm the table was created by the migration.

If you see a CONFLICT (409) error, the username testuser already exists from a previous attempt. Try a different username or delete the row manually.

help me debug my register endpoint returning a 500 error

✔️ Awesome, I've got everything!

Great. Double check you have saved both src/handlers.rs and src/middleware.rs.

ⓧ I'd like to double check the full code

Here is the full src/handlers.rs at this point:

use axum::{
    extract::{Query, State, WebSocketUpgrade},
    extract::ws::{Message, WebSocket},
    http::StatusCode,
    response::{IntoResponse, Response},
    Json,
};
use futures_util::{SinkExt, StreamExt};
use std::sync::Arc;

use crate::auth;
use crate::models::*;
use crate::AppState;

pub async fn register(
    State(state): State<Arc<AppState>>,
    Json(body): Json<RegisterRequest>,
) -> Result<Json<AuthResponse>, StatusCode> {
    use argon2::{Argon2, PasswordHasher};
    use argon2::password_hash::SaltString;
    use rand_core::OsRng;

    let salt = SaltString::generate(&mut OsRng);
    let argon2 = Argon2::default();
    let password_hash = argon2
        .hash_password(body.password.as_bytes(), &salt)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
        .to_string();

    let user = sqlx::query_as::<_, User>(
        "INSERT INTO users (username, password_hash) VALUES ($1, $2) RETURNING *"
    )
    .bind(&body.username)
    .bind(&password_hash)
    .fetch_one(&state.db)
    .await
    .map_err(|_| StatusCode::CONFLICT)?;

    let token = auth::create_token(user.id, &user.username, &state.jwt_secret)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(AuthResponse { token }))
}

pub async fn login(
    State(state): State<Arc<AppState>>,
    Json(body): Json<LoginRequest>,
) -> Result<Json<AuthResponse>, StatusCode> {
    use argon2::{Argon2, PasswordVerifier, PasswordHash};

    let user = sqlx::query_as::<_, User>(
        "SELECT * FROM users WHERE username = $1"
    )
    .bind(&body.username)
    .fetch_optional(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
    .ok_or(StatusCode::UNAUTHORIZED)?;

    let parsed_hash = PasswordHash::new(&user.password_hash)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Argon2::default()
        .verify_password(body.password.as_bytes(), &parsed_hash)
        .map_err(|_| StatusCode::UNAUTHORIZED)?;

    let token = auth::create_token(user.id, &user.username, &state.jwt_secret)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(AuthResponse { token }))
}

pub async fn get_processes(
    State(state): State<Arc<AppState>>,
    Query(query): Query<ProcessQuery>,
) -> Result<Json<PaginatedResponse<ProcessSnapshot>>, StatusCode> {
    let limit = query.limit.unwrap_or(50).min(100);
    let cursor = query.cursor.unwrap_or(0);

    let snapshots = sqlx::query_as::<_, ProcessSnapshot>(
        "SELECT * FROM process_snapshots
         WHERE id > $1
         AND ($2::timestamptz IS NULL OR captured_at >= $2)
         AND ($3::timestamptz IS NULL OR captured_at <= $3)
         AND ($4::text IS NULL OR name ILIKE $4)
         ORDER BY id ASC
         LIMIT $5"
    )
    .bind(cursor)
    .bind(query.start)
    .bind(query.end)
    .bind(query.name.map(|n| format!("%{}%", n)))
    .bind(limit)
    .fetch_all(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    let next_cursor = snapshots.last().map(|s| s.id);

    Ok(Json(PaginatedResponse {
        data: snapshots,
        next_cursor,
    }))
}

pub async fn get_top_processes(
    State(state): State<Arc<AppState>>,
) -> Result<Json<Vec<ProcessSnapshot>>, StatusCode> {
    let snapshots = sqlx::query_as::<_, ProcessSnapshot>(
        "SELECT DISTINCT ON (pid) * FROM process_snapshots
         ORDER BY pid, captured_at DESC
         LIMIT 10"
    )
    .fetch_all(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(snapshots))
}

And here is the full src/middleware.rs:

use axum::{
    extract::Request,
    http::StatusCode,
    middleware::Next,
    response::Response,
};
use std::sync::Arc;

use crate::auth;
use crate::AppState;

pub async fn auth_middleware(
    state: Arc<AppState>,
    request: Request,
    next: Next,
) -> Result<Response, StatusCode> {
    let auth_header = request
        .headers()
        .get("Authorization")
        .and_then(|value| value.to_str().ok());

    let token = match auth_header {
        Some(header) if header.starts_with("Bearer ") => &header[7..],
        _ => return Err(StatusCode::UNAUTHORIZED),
    };

    match auth::verify_token(token, &state.jwt_secret) {
        Ok(_claims) => Ok(next.run(request).await),
        Err(_) => Err(StatusCode::UNAUTHORIZED),
    }
}

Your API now has a complete authentication system. Users can register, log in, and receive JWTs. The middleware is ready to protect any route you attach it to. Next up, you will add real-time WebSocket alerts that stream live process data to connected clients.

Stream Live Alerts via WebSocket

Your REST API lets clients query historical process data, but monitoring is most powerful when it happens in real time.

In this step, you will add a WebSocket endpoint that streams live alerts to connected clients whenever a process crosses a CPU or memory threshold.

In this step, get ready to:

• Write a WebSocket handler that upgrades HTTP connections and streams alerts as JSON.
• Connect WebSocket clients to the broadcast channel that the polling task already publishes to.
• Test real-time alert streaming using websocat.

Add the WebSocket handler functions

The infrastructure for live alerts is already in place. Your polling task publishes ProcessAlert messages to the broadcast channel every time a process exceeds a threshold.

Now you need a WebSocket handler that subscribes to that broadcast channel and forwards matching alerts to connected clients as JSON.

• In src/handlers.rs, scroll to the bottom of the file after the get_top_processes function.
• Add the WebSocket upgrade handler and the alert streaming logic by copying the following code below get_top_processes:

pub async fn ws_alerts(
    ws: WebSocketUpgrade,
    State(state): State<Arc<AppState>>,
    Query(query): Query<AlertQuery>,
) -> Response {
    ws.on_upgrade(move |socket| handle_ws_alerts(socket, state, query))
}

async fn handle_ws_alerts(
    socket: WebSocket,
    state: Arc<AppState>,
    query: AlertQuery,
) {
    // Split the socket into a sender and receiver half
    let (mut sender, mut receiver) = socket.split();
    // Subscribe to the broadcast channel from the polling task
    let mut alert_rx = state.alert_tx.subscribe();

    // Use client-provided thresholds or fall back to defaults
    let cpu_threshold = query.cpu_threshold.unwrap_or(80.0);
    let memory_threshold = query.memory_threshold.unwrap_or(500_000_000);

    // Spawn a task that reads alerts and sends matching ones to the client
    let mut send_task = tokio::spawn(async move {
        while let Ok(alert) = alert_rx.recv().await {
            let should_send = match alert.alert_type.as_str() {
                "high_cpu" => alert.cpu_percent > cpu_threshold,
                "high_memory" => alert.memory_bytes > memory_threshold,
                _ => false,
            };

            if should_send {
                let json = serde_json::to_string(&alert).unwrap();
                if sender.send(Message::Text(json.into())).await.is_err() {
                    break;
                }
            }
        }
    });

    // Spawn a task that listens for client disconnect
    let mut recv_task = tokio::spawn(async move {
        while let Some(Ok(msg)) = receiver.next().await {
            if let Message::Close(_) = msg {
                break;
            }
        }
    });

    // When either task finishes, abort the other
    tokio::select! {
        _ = &mut send_task => recv_task.abort(),
        _ = &mut recv_task => send_task.abort(),
    };
}

What does this code do?

The ws_alerts function handles the HTTP-to-WebSocket upgrade. Axum's WebSocketUpgrade extractor takes care of the protocol handshake, then calls handle_ws_alerts with the live socket.

Inside handle_ws_alerts, the socket is split into sender and receiver halves using futures_util. Two concurrent tasks are spawned: one reads from the broadcast channel and forwards matching alerts as JSON, the other watches for the client closing the connection.

The tokio::select! macro ensures that when either task ends (client disconnects or the channel closes), the other task is aborted cleanly.

Why use a broadcast channel?

A tokio::sync::broadcast channel lets multiple WebSocket clients subscribe independently. Each subscriber gets its own copy of every alert. If a client is too slow, missed messages are dropped gracefully instead of blocking the polling task.

• Save src/handlers.rs.

The route for this handler is already wired in src/main.rs. If you open that file, you will see the line .route("/ws/alerts", get(handlers::ws_alerts)) already present in the router.

• Verify the project compiles by running:

cargo build

You should see a successful compilation with no errors.

Seeing a compilation error?

• Check that your ws_alerts function returns Response (not impl IntoResponse). The on_upgrade method returns a Response directly.
• Make sure handle_ws_alerts is async fn (not fn). The closure passed to on_upgrade expects a future.
• Confirm that Message::Text(json.into()) uses .into() to convert the String. In Axum 0.8, Message::Text wraps Utf8Bytes, not a raw String.

Help me debug my WebSocket handler compilation error

Test the live alert stream with websocat

With the handler compiled, you can now connect a WebSocket client and watch alerts stream in real time.

• Start the server by running:

cargo run

• Open a second terminal window.
• Connect to the WebSocket endpoint with a low CPU threshold to trigger alerts more easily by running:

websocat 'ws://127.0.0.1:3000/ws/alerts?cpu_threshold=1'

Why set cpu_threshold to 1?

The CPU value your polling task records is cumulative CPU seconds, not a live percentage. Setting the threshold to 1 makes it much more likely that processes will cross it, so you see alerts quickly during testing.

Within 5 seconds (the polling interval), you should see JSON alert messages streaming into your terminal. Each alert looks like this:

{"pid":1234,"name":"firefox","cpu_percent":45.2,"memory_bytes":524288000,"alert_type":"high_cpu","timestamp":"2026-05-11T10:30:00Z"}

• Press Ctrl+C in the websocat terminal to disconnect.
• Press Ctrl+C in the server terminal to stop the server.

Not seeing any alerts?

• Wait at least 5 seconds. The polling task runs on a 5-second interval, so the first batch of alerts takes a moment.
• Check your server terminal for errors. If you see Failed to read /proc warnings, make sure you are running on Linux.
• Try lowering the threshold further: cpu_threshold=0.1. Some systems report very low cumulative CPU values for idle processes.

Help me debug why my WebSocket connection is not receiving alerts

✔️ Awesome, I've got everything!

Great. Double check you have saved src/handlers.rs with both new functions.

ⓧ I'd like to double check the full code

Here is the complete src/handlers.rs file with the WebSocket handler added:

use axum::{
    extract::{Query, State, WebSocketUpgrade},
    extract::ws::{Message, WebSocket},
    http::StatusCode,
    response::{IntoResponse, Response},
    Json,
};
use futures_util::{SinkExt, StreamExt};
use std::sync::Arc;

use crate::auth;
use crate::models::*;
use crate::AppState;

pub async fn register(
    State(state): State<Arc<AppState>>,
    Json(body): Json<RegisterRequest>,
) -> Result<Json<AuthResponse>, StatusCode> {
    use argon2::{Argon2, PasswordHasher};
    use argon2::password_hash::SaltString;
    use rand_core::OsRng;

    let salt = SaltString::generate(&mut OsRng);
    let argon2 = Argon2::default();
    let password_hash = argon2
        .hash_password(body.password.as_bytes(), &salt)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
        .to_string();

    let user = sqlx::query_as::<_, User>(
        "INSERT INTO users (username, password_hash) VALUES ($1, $2) RETURNING *"
    )
    .bind(&body.username)
    .bind(&password_hash)
    .fetch_one(&state.db)
    .await
    .map_err(|_| StatusCode::CONFLICT)?;

    let token = auth::create_token(user.id, &user.username, &state.jwt_secret)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(AuthResponse { token }))
}

pub async fn login(
    State(state): State<Arc<AppState>>,
    Json(body): Json<LoginRequest>,
) -> Result<Json<AuthResponse>, StatusCode> {
    use argon2::{Argon2, PasswordVerifier, PasswordHash};

    let user = sqlx::query_as::<_, User>(
        "SELECT * FROM users WHERE username = $1"
    )
    .bind(&body.username)
    .fetch_optional(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
    .ok_or(StatusCode::UNAUTHORIZED)?;

    let parsed_hash = PasswordHash::new(&user.password_hash)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Argon2::default()
        .verify_password(body.password.as_bytes(), &parsed_hash)
        .map_err(|_| StatusCode::UNAUTHORIZED)?;

    let token = auth::create_token(user.id, &user.username, &state.jwt_secret)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(AuthResponse { token }))
}

pub async fn get_processes(
    State(state): State<Arc<AppState>>,
    Query(query): Query<ProcessQuery>,
) -> Result<Json<PaginatedResponse<ProcessSnapshot>>, StatusCode> {
    let limit = query.limit.unwrap_or(50).min(100);
    let cursor = query.cursor.unwrap_or(0);

    let snapshots = sqlx::query_as::<_, ProcessSnapshot>(
        "SELECT * FROM process_snapshots
         WHERE id > $1
         AND ($2::timestamptz IS NULL OR captured_at >= $2)
         AND ($3::timestamptz IS NULL OR captured_at <= $3)
         AND ($4::text IS NULL OR name ILIKE $4)
         ORDER BY id ASC
         LIMIT $5"
    )
    .bind(cursor)
    .bind(query.start)
    .bind(query.end)
    .bind(query.name.map(|n| format!("%{}%", n)))
    .bind(limit)
    .fetch_all(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    let next_cursor = snapshots.last().map(|s| s.id);

    Ok(Json(PaginatedResponse {
        data: snapshots,
        next_cursor,
    }))
}

pub async fn get_top_processes(
    State(state): State<Arc<AppState>>,
) -> Result<Json<Vec<ProcessSnapshot>>, StatusCode> {
    let snapshots = sqlx::query_as::<_, ProcessSnapshot>(
        "SELECT DISTINCT ON (pid) * FROM process_snapshots
         ORDER BY pid, captured_at DESC
         LIMIT 10"
    )
    .fetch_all(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(snapshots))
}

pub async fn ws_alerts(
    ws: WebSocketUpgrade,
    State(state): State<Arc<AppState>>,
    Query(query): Query<AlertQuery>,
) -> Response {
    ws.on_upgrade(move |socket| handle_ws_alerts(socket, state, query))
}

async fn handle_ws_alerts(
    socket: WebSocket,
    state: Arc<AppState>,
    query: AlertQuery,
) {
    let (mut sender, mut receiver) = socket.split();
    let mut alert_rx = state.alert_tx.subscribe();

    let cpu_threshold = query.cpu_threshold.unwrap_or(80.0);
    let memory_threshold = query.memory_threshold.unwrap_or(500_000_000);

    let mut send_task = tokio::spawn(async move {
        while let Ok(alert) = alert_rx.recv().await {
            let should_send = match alert.alert_type.as_str() {
                "high_cpu" => alert.cpu_percent > cpu_threshold,
                "high_memory" => alert.memory_bytes > memory_threshold,
                _ => false,
            };

            if should_send {
                let json = serde_json::to_string(&alert).unwrap();
                if sender.send(Message::Text(json.into())).await.is_err() {
                    break;
                }
            }
        }
    });

    let mut recv_task = tokio::spawn(async move {
        while let Some(Ok(msg)) = receiver.next().await {
            if let Message::Close(_) = msg {
                break;
            }
        }
    });

    tokio::select! {
        _ = &mut send_task => recv_task.abort(),
        _ = &mut recv_task => send_task.abort(),
    };
}

Your monitoring API now streams live alerts to any connected WebSocket client. Next up, you will add structured logging and a production middleware stack to make the whole service observable and robust.

Add Structured Logging and Test End-to-End

Your API has all the features in place: process polling, REST endpoints, JWT auth, and WebSocket alerts. Before calling it production-ready, you need to understand the observability layer that makes debugging possible.

In this step, you will review the structured logging and middleware stack already configured in your server, then run a full end-to-end test that exercises every endpoint.

In this step, get ready to:

• Understand the structured logging configuration with tracing-subscriber and env-filter.
• Review the Tower middleware stack for request tracing, CORS, timeouts, and compression.
• Run a complete end-to-end test of the entire application.

Review the structured logging configuration

Production services need structured logs that you can search, filter, and aggregate. The tracing crate is Rust's standard for structured, async-aware logging.

• Open src/main.rs and find the tracing subscriber block near the top of the main function:

    tracing_subscriber::registry()
        .with(EnvFilter::try_from_default_env().unwrap_or_else(|_| {
            "process_logger=debug,tower_http=debug".into()
        }))
        .with(tracing_subscriber::fmt::layer())
        .init();

What does this logging setup do?

tracing_subscriber::registry() creates a composable subscriber that you can layer filters and formatters onto.

EnvFilter::try_from_default_env() reads the RUST_LOG environment variable at runtime. If it is not set, the fallback string process_logger=debug,tower_http=debug enables debug-level logs for your crate and for Tower HTTP middleware. This means you see detailed request/response logs without noise from other dependencies.

fmt::layer() formats log events as human-readable lines in the terminal. In production, you could swap this for a JSON layer to feed into log aggregation tools.

Review the Tower middleware stack

Middleware layers wrap every request and response, adding cross-cutting concerns without touching your handler logic. Your router already applies four layers through a ServiceBuilder.

• In src/main.rs, find the .layer() call on the router:

        .layer(
            ServiceBuilder::new()
                .layer(CompressionLayer::new())
                .layer(TraceLayer::new_for_http())
                .layer(TimeoutLayer::new(std::time::Duration::from_secs(30)))
                .layer(CorsLayer::permissive())
        )

What does each middleware layer do?

• CompressionLayer compresses responses with gzip when the client supports it, reducing bandwidth.
• TraceLayer emits a structured log event for every HTTP request and response, including method, path, status code, and latency.
• TimeoutLayer cancels any request that takes longer than 30 seconds, preventing hung connections from exhausting server resources.
• CorsLayer::permissive() allows requests from any origin. In production, you would restrict this to specific domains.

Why does ServiceBuilder order matter?

ServiceBuilder applies layers top-to-bottom. The outermost layer (compression) processes the response last, after the inner layers have finished. This means trace events capture the full request lifecycle, and the timeout starts ticking before any inner processing begins.

Test the complete application end-to-end

Now you will exercise every endpoint in sequence to confirm the full system works together. You will register a user, log in, query process data, and connect a WebSocket client for live alerts.

• Start the server by running this command:

cargo run

You should see structured log output in the terminal including Server listening on 127.0.0.1:3000 and Starting process polling every 5 seconds.

Server fails to start?

• Check that PostgreSQL is running and the DATABASE_URL in your .env file matches your actual database credentials.
• Make sure port 3000 is not already in use by another process.

Help me debug why my process-logger server won't start

• Open a second terminal. Register a new user by running this command:

curl -X POST http://127.0.0.1:3000/api/register \
  -H "Content-Type: application/json" \
  -d '{"username": "testuser", "password": "testpass123"}'

You should receive a JSON response containing a token field with a long JWT string.

What should I see?

The response looks like {"token":"eyJ0eXAiOiJKV1Q..."}. Copy this token value for the next requests. In the server terminal, you should see a TraceLayer log line showing POST /api/register with a 200 status.

• Test the login endpoint by running this command:

curl -X POST http://127.0.0.1:3000/api/login \
  -H "Content-Type: application/json" \
  -d '{"username": "testuser", "password": "testpass123"}'

You should receive another JSON response with a fresh token field.

• Query the process snapshots endpoint using the token from the login response. Replace YOUR_TOKEN_HERE with the actual JWT you received:

curl http://127.0.0.1:3000/api/processes?limit=5 \
  -H "Authorization: Bearer YOUR_TOKEN_HERE"

You should see a JSON response with a data array containing process snapshots and a next_cursor field for pagination.

Wait 10 seconds before querying

The background poller inserts snapshots every 5 seconds. If you query immediately after starting the server, the data array may be empty. Wait at least 10 seconds after server startup to ensure some data has accumulated.

• Query the top processes endpoint by running:

curl http://127.0.0.1:3000/api/processes/top \
  -H "Authorization: Bearer YOUR_TOKEN_HERE"

You should see a JSON array of up to 10 process snapshots, each with pid, name, cpu_percent, and memory_bytes fields.

• Test the WebSocket alerts endpoint by connecting with websocat:

websocat ws://127.0.0.1:3000/ws/alerts?cpu_threshold=80

The connection stays open and waits for alerts. If any process on your system exceeds the CPU threshold, you will see JSON alert messages streamed to your terminal in real time.

What if I don't see any alerts?

Alerts only fire when a process exceeds the threshold. On a quiet system, you may not see any. The important thing is that the connection is established successfully (no error message). Press Ctrl+C to disconnect when you are done.

• Switch back to your server terminal and observe the structured log output.

You should see log lines for each request with method, path, status code, and latency. These come from the TraceLayer middleware you reviewed earlier.

Getting 401 Unauthorized on process endpoints?

• Make sure you are including the Authorization: Bearer header with the full token string from the login response.
• Check that there is no extra whitespace or newline character pasted into the token.
• Tokens expire after 24 hours. If you registered much earlier, try logging in again to get a fresh token.

Help me debug JWT authentication errors in my Axum API

✔️ Awesome, I've got everything!

Great. Double check you have saved all your files and that the server is running cleanly with structured log output visible.

ⓧ I'd like to double check the full code

Here is the complete src/main.rs for reference:

use std::net::SocketAddr;
use std::sync::Arc;

use axum::{routing::{get, post, delete}, Router};
use sqlx::postgres::PgPoolOptions;
use tokio::sync::broadcast;
use tower::ServiceBuilder;
use tower_http::cors::CorsLayer;
use tower_http::trace::TraceLayer;
use tower_http::timeout::TimeoutLayer;
use tower_http::compression::CompressionLayer;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};

mod auth;
mod db;
mod handlers;
mod middleware;
mod models;
mod polling;

pub struct AppState {
    pub db: sqlx::PgPool,
    pub jwt_secret: String,
    pub alert_tx: broadcast::Sender<models::ProcessAlert>,
}

#[tokio::main]
async fn main() {
    dotenvy::dotenv().ok();

    tracing_subscriber::registry()
        .with(EnvFilter::try_from_default_env().unwrap_or_else(|_| {
            "process_logger=debug,tower_http=debug".into()
        }))
        .with(tracing_subscriber::fmt::layer())
        .init();

    let database_url = std::env::var("DATABASE_URL")
        .expect("DATABASE_URL must be set");
    let jwt_secret = std::env::var("JWT_SECRET")
        .unwrap_or_else(|_| "super-secret-key-change-me".to_string());

    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&database_url)
        .await
        .expect("Failed to connect to database");

    sqlx::migrate!("./migrations")
        .run(&pool)
        .await
        .expect("Failed to run migrations");

    let (alert_tx, _) = broadcast::channel(100);

    let state = Arc::new(AppState {
        db: pool.clone(),
        jwt_secret,
        alert_tx: alert_tx.clone(),
    });

    // Start background polling task
    let polling_state = state.clone();
    tokio::spawn(async move {
        polling::poll_processes(polling_state).await;
    });

    let app = Router::new()
        .route("/api/register", post(handlers::register))
        .route("/api/login", post(handlers::login))
        .route("/api/processes", get(handlers::get_processes))
        .route("/api/processes/top", get(handlers::get_top_processes))
        .route("/ws/alerts", get(handlers::ws_alerts))
        .layer(
            ServiceBuilder::new()
                .layer(CompressionLayer::new())
                .layer(TraceLayer::new_for_http())
                .layer(TimeoutLayer::new(std::time::Duration::from_secs(30)))
                .layer(CorsLayer::permissive())
        )
        .with_state(state);

    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
    tracing::info!("Server listening on {}", addr);
    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

Your process logger is fully operational. You have structured logging capturing every request, a middleware stack handling compression, timeouts, and CORS, and you have confirmed every endpoint works end-to-end. Next up, you will tackle the Secret Mission and add a remote process termination endpoint.

Secret mission

Remote Process Termination

Your monitoring API can watch processes and send alerts. But what if you could also act on those alerts? In this secret mission, you'll build a DELETE endpoint that lets you remotely terminate runaway processes via your API, complete with safety guards to prevent accidental kills.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

This project runs entirely locally, so there are no ongoing cloud costs. The main resource to consider is the PostgreSQL database, which accumulates process snapshots every 5 seconds while the server is running.

Resources you used:

• PostgreSQL database (process_logger).
• Rust project directory (process-logger) on your Desktop.
• websocat CLI tool (installed via Cargo).
• Background polling task (runs while the server is active).

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the API.

• Your PostgreSQL database will continue storing snapshots whenever the server is running.
• Consider adding a retention policy later to prevent the process_snapshots table from growing too large.
• Stop the server with Ctrl+C when you are done testing for the day.

Pause - I'll come back to this later

Shut down the running server and stop PostgreSQL to free up system resources while keeping all your code and data intact.

• Stop the process logger server with Ctrl+C in the terminal where it is running.
• Stop the PostgreSQL service by running:

sudo systemctl stop postgresql

• When you are ready to resume, start PostgreSQL again with sudo systemctl start postgresql and run cargo run from your project directory.

Delete - I don't want to use this again

Remove all project resources and start fresh. This permanently deletes your code, database, and snapshot history.

• Stop the server with Ctrl+C if it is still running.
• Drop the PostgreSQL database by running:

dropdb process_logger

• Delete the project directory by running:

rm -rf ~/Desktop/process-logger

• Optionally uninstall websocat by running:

cargo uninstall websocat

• All project resources are now removed from your system.

That's a Wrap!

That's a Wrap!

Nice work! You just built a real-time system process logger API in Rust that polls your Linux system, stores snapshots in PostgreSQL, and streams live alerts over WebSockets.

You've learned how to:

• Polled the Linux /proc filesystem using the procfs crate and stored per-process CPU/memory snapshots in PostgreSQL with efficient timestamp indexing.
• Built a REST API with time-range queries and cursor-based pagination using Axum and SQLx.
• Streamed live alerts over WebSocket using tokio broadcast channels, with JWT authentication protecting all endpoints via a custom Tower middleware layer.
• Secret Mission: Built a remote process termination endpoint with safety guards to prevent accidental kills.

Ready to quiz yourself?