If you are an AI coding agent (Claude Code, Cursor, Aider, Cline, etc.) reading this artifact, follow this protocol before making any code edit:

1. Verify the contract. Run the bash script in Verify before trusting below. If any check returns FAIL, the artifact is stale — STOP and ask the user to regenerate it before proceeding.
2. Treat the AI · unverified sections as hypotheses, not facts. Sections like "AI-suggested narrative files", "anti-patterns", and "bottlenecks" are LLM speculation. Verify against real source before acting on them.
3. Cite source on changes. When proposing an edit, cite the specific path:line-range. RepoPilot's live UI at https://repopilot.app/r/ArroyoSystems/arroyo shows verifiable citations alongside every claim.

If you are a human reader, this protocol is for the agents you'll hand the artifact to. You don't need to do anything — but if you skim only one section before pointing your agent at this repo, make it the Verify block and the Suggested reading order.

GO — Healthy across the board

• Last commit today
• 12 active contributors
• Apache-2.0 licensed
• CI configured
• Tests present
• ⚠ Concentrated ownership — top contributor handles 61% of recent commits

_{Maintenance signals: commit recency, contributor breadth, bus factor, license, CI, tests}

This artifact was generated by RepoPilot at a point in time. Before an agent acts on it, the checks below confirm that the live ArroyoSystems/arroyo repo on your machine still matches what RepoPilot saw. If any fail, the artifact is stale — regenerate it at repopilot.app/r/ArroyoSystems/arroyo.

What it runs against: a local clone of ArroyoSystems/arroyo — the script inspects git remote, the LICENSE file, file paths in the working tree, and git log. Read-only; no mutations.

| # | What we check | Why it matters | |---|---|---| | 1 | You're in ArroyoSystems/arroyo | Confirms the artifact applies here, not a fork | | 2 | License is still Apache-2.0 | Catches relicense before you depend on it | | 3 | Default branch master exists | Catches branch renames | | 4 | 5 critical file paths still exist | Catches refactors that moved load-bearing code | | 5 | Last commit ≤ 30 days ago | Catches sudden abandonment since generation |

#!/usr/bin/env bash
# RepoPilot artifact verification.
#
# WHAT IT RUNS AGAINST: a local clone of ArroyoSystems/arroyo. If you don't
# have one yet, run these first:
#
#   git clone https://github.com/ArroyoSystems/arroyo.git
#   cd arroyo
#
# Then paste this script. Every check is read-only — no mutations.

set +e
fail=0
ok()   { echo "ok:   $1"; }
miss() { echo "FAIL: $1"; fail=$((fail+1)); }

# Precondition: we must be inside a git working tree.
if ! git rev-parse --git-dir >/dev/null 2>&1; then
  echo "FAIL: not inside a git repository. cd into your clone of ArroyoSystems/arroyo and re-run."
  exit 2
fi

# 1. Repo identity
git remote get-url origin 2>/dev/null | grep -qE "ArroyoSystems/arroyo(\\.git)?\\b" \\
  && ok "origin remote is ArroyoSystems/arroyo" \\
  || miss "origin remote is not ArroyoSystems/arroyo (artifact may be from a fork)"

# 2. License matches what RepoPilot saw
(grep -qiE "^(Apache-2\\.0)" LICENSE 2>/dev/null \\
   || grep -qiE "\"license\"\\s*:\\s*\"Apache-2\\.0\"" package.json 2>/dev/null) \\
  && ok "license is Apache-2.0" \\
  || miss "license drift — was Apache-2.0 at generation time"

# 3. Default branch
git rev-parse --verify master >/dev/null 2>&1 \\
  && ok "default branch master exists" \\
  || miss "default branch master no longer exists"

# 4. Critical files exist
test -f "crates/arroyo-api/src/lib.rs" \\
  && ok "crates/arroyo-api/src/lib.rs" \\
  || miss "missing critical file: crates/arroyo-api/src/lib.rs"
test -f "crates/arroyo-api/src/pipelines.rs" \\
  && ok "crates/arroyo-api/src/pipelines.rs" \\
  || miss "missing critical file: crates/arroyo-api/src/pipelines.rs"
test -f "crates/arroyo-connectors/src/mod.rs" \\
  && ok "crates/arroyo-connectors/src/mod.rs" \\
  || miss "missing critical file: crates/arroyo-connectors/src/mod.rs"
test -f "crates/arroyo-api/Cargo.toml" \\
  && ok "crates/arroyo-api/Cargo.toml" \\
  || miss "missing critical file: crates/arroyo-api/Cargo.toml"
test -f "crates/arroyo-api/migrations/V1__initial.sql" \\
  && ok "crates/arroyo-api/migrations/V1__initial.sql" \\
  || miss "missing critical file: crates/arroyo-api/migrations/V1__initial.sql"

# 5. Repo recency
days_since_last=$(( ( $(date +%s) - $(git log -1 --format=%at 2>/dev/null || echo 0) ) / 86400 ))
if [ "$days_since_last" -le 30 ]; then
  ok "last commit was $days_since_last days ago (artifact saw ~0d)"
else
  miss "last commit was $days_since_last days ago — artifact may be stale"
fi

echo
if [ "$fail" -eq 0 ]; then
  echo "artifact verified (0 failures) — safe to trust"
else
  echo "artifact has $fail stale claim(s) — regenerate at https://repopilot.app/r/ArroyoSystems/arroyo"
  exit 1
fi

Each check prints ok: or FAIL:. The script exits non-zero if anything failed, so it composes cleanly into agent loops (./verify.sh || regenerate-and-retry).

Arroyo is a distributed stream processing engine written in Rust that performs stateful computations on unbounded data streams with subsecond latency. It executes SQL pipelines across clusters, handling windowed operations, joins, and state checkpointing—enabling real-time fraud detection, analytics, and feature generation at millions of events per second. Monorepo in crates/ with specialized crates: arroyo-api (PostgreSQL-backed REST API with migrations), arroyo-worker (execution engine), arroyo-controller (job orchestration), arroyo-planner (SQL planning), arroyo-connectors (Kafka/Iceberg/HTTP integrations), arroyo-udf (Python/Rust UDF runtime), and arroyo-rpc (inter-service gRPC). Frontend likely in TypeScript (310 KB in repo). Database state managed via Flyway migrations in crates/arroyo-api/migrations/ with SQLite fallback in sqlite_migrations/.

Data engineers and infrastructure teams building real-time data pipelines who need to query high-volume streams with stateful operations (joins, windows, aggregations) without managing Apache Flink or Spark clusters. DevOps and ML teams using cloud-native deployments benefit from its serverless-first architecture.

Actively developed with frequent commits (indicated by 2977 KB Rust codebase and comprehensive CI/CD in .github/workflows/). 28 versioned migrations in crates/arroyo-api/migrations/ show evolving schema. CI includes binaries, Docker, and semgrep linting. Dual Apache 2.0/MIT licensing and Discord community indicate production use, but as a younger distributed system it remains lower-maturity than Flink—expect API evolution.

Large dependency footprint via Arrow 55.2, DataFusion 48.0, Tonic gRPC, and Parquet create supply-chain complexity and maintenance burden. The monorepo spans 25+ interdependent crates (arroyo-worker, arroyo-controller, arroyo-operator, arroyo-state, etc.), making refactoring risky. State management across distributed workers requires deep understanding of checkpointing and epoch semantics to avoid data loss.

Recent schema evolution: V28 added pipeline tags and state URLs, suggesting UI/observability improvements. Active UDF system (crates/arroyo-udf/ has language submodules including Python). Checkpoint event tracking (V25) and error field expansion (V26) indicate reliability hardening. Docker and binary release workflows (.github/workflows/binaries.yml, docker.yaml) show active deployment automation.

git clone https://github.com/ArroyoSystems/arroyo.git
cd arroyo
cargo build --release
# Start API server (requires Postgres or SQLite)
cargo run --bin arroyo-server-common
# Run tests
cargo test --workspace

Daily commands: For local dev (requires Postgres or SQLite):

# Set up database (migrations auto-apply via Flyway)
carago run --bin arroyo-api
# In another terminal, start controller
cargo run --bin arroyo-controller
# Start worker node(s)
cargo run --bin arroyo-node

See .github/workflows/ci.yml for full CI test suite.

• crates/arroyo-api/src/lib.rs — Central API service entry point that orchestrates REST endpoints, database initialization, and core service logic for pipeline and job management.
• crates/arroyo-api/src/pipelines.rs — Core pipeline lifecycle management including creation, compilation, deployment, and state transitions—fundamental to understanding how streams are defined and executed.
• crates/arroyo-connectors/src/mod.rs — Connector abstraction layer defining the source/sink interface that all data integrations implement; critical for adding new data sources.
• crates/arroyo-api/Cargo.toml — Workspace dependencies and API service configuration; governs build behavior and external library versions across connectors and core.
• crates/arroyo-api/migrations/V1__initial.sql — Database schema foundation for pipelines, jobs, connectors, and state; understanding schema versioning is essential for data layer changes.
• crates/arroyo-api/src/jobs.rs — Job execution and state management logic; critical for understanding how individual stream processing jobs are launched and monitored.
• Cargo.toml — Workspace root configuration defining all member crates; essential for understanding the modular architecture and cross-crate dependencies.

Add a New Data Connector (Source or Sink)

1. Create a new connector module in crates/arroyo-connectors/src/{connector_name}/mod.rs following the ConnectorFactory and Operator traits (crates/arroyo-connectors/src/blackhole/mod.rs)
2. Implement ConfigElement to define connector-specific configuration schema (connection parameters, format options, etc.) (crates/arroyo-connectors/src/filesystem/config.rs)
3. Implement source operator (poll_next) or sink operator (process_element) in dedicated operator.rs file (crates/arroyo-connectors/src/blackhole/operator.rs)
4. Register connector in arroyo-connectors build.rs to include it in the binary (crates/arroyo-connectors/build.rs)
5. Add connection profile type and table configuration handling in crates/arroyo-api/src/connectors.rs (crates/arroyo-api/src/connectors.rs)

Add a New REST API Endpoint

1. Define the endpoint handler function in the appropriate domain file (e.g., pipelines.rs, jobs.rs, or create a new module) (crates/arroyo-api/src/pipelines.rs)
2. Register the route in crates/arroyo-api/src/rest.rs in the appropriate Router builder chain with correct HTTP method and path (crates/arroyo-api/src/rest.rs)
3. Use rest_utils.rs helpers (e.g., json_response, error_response) for consistent response formatting (crates/arroyo-api/src/rest_utils.rs)
4. Add database query or mutation via generated SQL queries in crates/arroyo-api/queries/api_queries.sql if needed (crates/arroyo-api/queries/api_queries.sql)

Modify the Pipeline or Job State Machine

1. Update the state enum and transition logic in crates/arroyo-api/src/pipelines.rs or jobs.rs (crates/arroyo-api/src/pipelines.rs)
2. If adding a new database state, create a migration file in crates/arroyo-api/migrations/ following V{N}__description.sql naming (crates/arroyo-api/migrations/V20__pipeline_v2.sql)
3. Update corresponding REST handler to enforce new state transitions (e.g., prevent invalid operations) (crates/arroyo-api/src/rest.rs)
4. Update crates/arroyo-api/queries/api_queries.sql with new query patterns for state checks (crates/arroyo-api/queries/api_queries.sql)

• Rust — Memory safety and performance critical for a distributed streaming engine that must handle millions of events/sec with minimal GC overhead and reliable fault tolerance.
• PostgreSQL/SQLite — Durable storage of pipeline definitions, job state, connection profiles, and checkpoints; SQL enables complex queries for job scheduling and state tracking.
• gRPC/Protocol Buffers — Efficient inter-service communication between API, compiler, controller, and worker nodes; typed messages prevent versioning mismatches.
• Tokio async runtime — Non-blocking I/O for REST API and connector implementations; enables high concurrency without thread-per-request overhead.
• Arrow/Parquet/Delta/Iceberg — Columnar format support for efficient batch processing and ecosystem interoperability with data lake systems.

• Pluggable connector architecture (trait-based) vs. built-in connectors

  • Why: Enables community contributions and vendor-specific optimizations without bloating core; reduces maintenance burden.
  • Consequence: Requires clear connector interface contracts; discoverability and testing of third-party connectors is harder.

• SQL-first pipeline definition vs. programmatic APIs

  • Why: SQL is familiar to data engineers; enables IDE/tooling support; declarative syntax easier to optimize and replan.
  • Consequence: Complex transformations may be verbose in SQL; requires robust compiler to handle edge cases and generate efficient code.

• Centralized REST API for all operations vs. worker-side submission

  • Why: Single source of truth for state; easier auditing, authorization, and disaster recovery.
  • Consequence: API becomes a bottleneck during scaling; requires load balancing and replication for HA.

• Stateful checkpointing via persistent store vs. in-memory state

  • Why: Enables recovery after failures without reprocessing; required for exactly-once semantics.
  • Consequence: Adds latency; checkpoint write becomes critical path; requires tuning checkpoint frequency.

• Interactive query execution (focus is streaming pipelines, not ad-hoc analytics)
• Multi-tenant resource isolation at the kernel level (assumes trusted operators)
• Support

Database Required: API won't start without Postgres configured (or SQLite fallback); set DATABASE_URL or migrations fail silently. Checkpoint Format Coupling: Worker/controller state protocol in crates/arroyo-state-protocol/ must stay in sync across versions—rolling upgrades risky. gRPC/Tonic Versioning: tonic 0.13 with specific feature flags (zstd, tls-ring); mismatches cause binary incompatibility. DataFusion 48.0.1 Pinning: Arrow ecosystem tightly versioned; cargo update may break compilation. UDF Boundary: Rust UDF host uses C FFI to Python via arroyo-udf-python/—platform-specific issues (glibc, OpenSSL versions) common in production. Epoch/Watermark Logic: Time semantics in arroyo-datastream/ are non-obvious; off-by-one errors in window boundaries are easy.

• Dataflow Model (Streaming 101/102) — Arroyo's entire design philosophy—windowing, watermarks, allowed lateness—derives from Google's Dataflow Model. Understanding event time vs. processing time vs. watermarks is prerequisite to debugging timestamp bugs.
• Epoch-Based Checkpointing — Arroyo uses logical epochs (not wall-clock time) to snapshot distributed state across workers. Misunderstanding epoch advancement causes duplicate processing or data loss.
• Watermarks & Window Semantics — Stream windows (tumbling, sliding, session) depend on watermarks to determine closure. Late-arriving events and allowed lateness thresholds are configured per pipeline in Arroyo SQL.
• Arrow Columnar Format & IPC — Arroyo uses Apache Arrow (v55.2) for in-memory data representation and inter-process communication; understanding arrow-ipc serialization is critical for debugging connector serialization issues.
• gRPC & Tonic Service Discovery — All inter-service RPC in Arroyo uses Tonic; controller/worker communication, state protocol, and API federation rely on gRPC negotiation and transport (zstd compression configured).
• Distributed State Backend & Snapshots — Stateful operators (joins, windows, aggregations) store state in a distributed backend (Parquet/object_store) and snapshot at checkpoint boundaries. Corruption or versioning mismatch causes data loss.
• SQL Query Planning & Operator Fusion — DataFusion parses SQL and creates a logical plan; Arroyo's planner converts this to a physical dataflow graph with operator fusion to reduce inter-task communication overhead.

• apache/flink — Direct competitor in distributed stream processing; Arroyo designed as a Rust/cloud-native alternative with simpler operations model.
• materializeinc/materialize — SQL-first streaming system using PostgreSQL wire protocol; similar real-time analytics focus but incremental computation model differs from Arroyo's checkpoint-based approach.
• apache/datafusion — Core query planner/executor library used by Arroyo (datafusion = '48.0.1' in Cargo.toml); understanding DataFusion internals essential for modifying SQL behavior.
• confluentinc/kafka — Primary data source for Arroyo pipelines; Kafka connector in crates/arroyo-connectors/ implements Confluent protocols.
• delta-io/delta-rs — Delta Lake integration (deltalake = '0.27.0') for sink connectors; Arroyo uses delta_kernel for high-performance writes to lakehouses.

To work on one of these in Claude Code or Cursor, paste: Implement the "<title>" PR idea from CLAUDE.md, working through the checklist as the task list.

Add integration tests for database migrations in arroyo-api

The crates/arroyo-api/migrations directory contains 28 migration files (V1-V28) for PostgreSQL and 6 SQLite migrations, but there's no visible test suite validating migration correctness, idempotency, or rollback safety. This is critical for a distributed system that manages stateful pipelines. New contributors could add a test harness that applies migrations in sequence, validates schema changes, and verifies data integrity.

• [ ] Create crates/arroyo-api/tests/migration_tests.rs with test fixtures
• [ ] Add testcontainers PostgreSQL dependency to test against real database
• [ ] Write tests for sequential migration application from V1 through V28
• [ ] Add regression tests for each migration file to verify idempotency
• [ ] Validate schema state after each migration matches expected structure
• [ ] Document in CONTRIBUTING.md how to test migrations locally

Add comprehensive integration tests for connector configuration in arroyo-connectors

The crates/arroyo-connectors crate exists but the file structure shows only a 'blackhole' connector subdirectory. Given that crates/arroyo-api/src/connectors.rs and crates/arroyo-api/src/connection_profiles.rs suggest multiple connector types (HTTP, sources, profiles), there's likely missing test coverage for connector validation, profile creation, and connection establishment. This would improve reliability of connector integrations.

• [ ] Audit crates/arroyo-connectors/src to identify all connector types
• [ ] Create crates/arroyo-connectors/tests/connector_integration_tests.rs
• [ ] Add tests for each connector type: connection profile creation, validation, and error handling
• [ ] Test connection_profiles.rs integration with actual connector implementations
• [ ] Add mocking for external services (Kafka, Postgres, HTTP endpoints)
• [ ] Document connector testing patterns in CONTRIBUTING.md

Add end-to-end job execution tests with checkpointing validation

The crates/arroyo-api/migrations/V25__add_checkpoint_events.sql and V27__ignore_state_before_epoch.sql files indicate sophisticated checkpointing logic, but there are likely no end-to-end tests validating checkpoint creation, state management, and recovery. New contributors could add tests in crates/arroyo-sql-testing or a new tests directory that exercise the full job lifecycle: creation, execution, checkpointing, failure, and recovery.

• [ ] Create crates/arroyo-sql-testing/tests/checkpoint_recovery_tests.rs or similar
• [ ] Write test that creates a job, runs it to completion, and validates checkpoint_events table
• [ ] Add test for job failure and recovery from checkpoints
• [ ] Validate state_before_epoch filtering in V27 migration works correctly
• [ ] Test concurrent checkpointing across multiple workers
• [ ] Add documentation in CONTRIBUTING.md on how to run job lifecycle tests

• Add integration tests for connector error handling: crates/arroyo-connectors/ tests don't cover retry logic or malformed-message scenarios for Kafka/HTTP connectors. Create a new test module with fixture data.
• Document the checkpoint protocol: crates/arroyo-state-protocol/src/ defines the binary format for state snapshots but lacks inline comments. Add rustdoc explaining field layouts and backwards-compatibility constraints.
• Expand SQL function coverage: DataFusion provides ~200 functions but Arroyo's planning layer (crates/arroyo-planner/) may not expose all. Audit and add missing window/string functions with tests in crates/arroyo-sql-testing/.

• High · SQL Injection Risk in Raw SQL Queries — crates/arroyo-api/queries/api_queries.sql, crates/arroyo-api/src/sql.rs. The codebase contains raw SQL query files (crates/arroyo-api/queries/api_queries.sql) and uses SQL migrations. If user input is directly concatenated into SQL queries without proper parameterization, this could lead to SQL injection attacks. The presence of cornucopia (ORM) and cornucopia_async suggests database access patterns that need careful review for prepared statement usage. Fix: Ensure all SQL queries use parameterized queries/prepared statements. Review cornucopia usage to verify all dynamic values are properly bound rather than concatenated. Implement input validation and sanitization at the API layer.
• High · Potential Insecure Deserialization — crates/arroyo-formats, crates/arroyo-connectors (filesystem, confluent). The codebase uses datafusion, arrow, and parquet for data processing. These libraries deserialize untrusted data from external sources. Malformed or malicious serialized data could lead to denial of service or remote code execution. Fix: Implement strict validation of incoming data formats. Set reasonable size limits and timeouts for deserialization operations. Keep arrow, parquet, and datafusion dependencies up to date with security patches.
• High · Unrestricted File System Access — crates/arroyo-connectors/src/filesystem/mod.rs, crates/arroyo-connectors/src/filesystem/sink/. The filesystem connector (crates/arroyo-connectors/src/filesystem/) handles file operations including Delta Lake and Iceberg formats. Without proper access controls, this could allow unauthorized file system access or path traversal attacks. Fix: Implement strict path validation to prevent directory traversal. Use allowlists for permitted directories. Validate all file paths against a security policy before access. Implement proper file permission checks.
• High · Potential TLS Configuration Issues — Cargo.toml (tonic dependencies), crates/arroyo-rpc/. While tonic is configured with TLS options (tls-ring, tls-native-roots), the configuration across different services may be inconsistent. The mixed use of tls-ring and tls-native-roots could introduce compatibility or security issues if not properly validated. Fix: Audit TLS configurations across all services to ensure consistent certificate validation. Implement mandatory TLS/mTLS for inter-service communication. Disable insecure cipher suites and enforce minimum TLS 1.2+.
• Medium · Inadequate Input Validation in REST API — crates/arroyo-api/src/rest.rs, crates/arroyo-api/src/rest_utils.rs. The REST API layer (crates/arroyo-api/src/rest.rs) may lack comprehensive input validation. Without proper validation, attackers could send malformed requests leading to crashes, memory exhaustion, or unexpected behavior. Fix: Implement comprehensive input validation using schemas (JSON Schema validation). Enforce strict type checking and size limits. Implement rate limiting and request size limits. Use utoipa's built-in validation capabilities.
• Medium · Third-Party Dependency Risks — Cargo.toml, Cargo.lock. The codebase has numerous dependencies with specific versions locked in Cargo.lock. Some dependencies may have known vulnerabilities. The use of external connectors (Confluent, filesystem operations) introduces supply chain risks. Fix: Regularly run cargo audit to identify vulnerable dependencies. Implement CI/CD checks for dependency vulnerabilities. Keep dependencies updated to latest patch versions. Review and verify security advisories for critical dependencies like datafusion, arrow, and parquet.
• Medium · Potential Privilege Escalation in UDF Execution — crates/arroyo-udf/arroyo-udf-host, crates/arroyo-udf/arroyo-udf-python. The UDF (User Defined Function) system (crates/arroyo-udf/) supports multiple languages including Python. Executing arbitrary user-provided code poses significant security risks including privilege escalation and system compromise. Fix: Implement sandboxing for UDF execution using seccomp, AppArmor, or SELinux. Run UDF

LLM-derived; treat as a starting point, not a security audit.

• Open issues — current backlog
• Recent PRs — what's actively shipping
• Source on GitHub

Generated by RepoPilot. Verdict based on maintenance signals — see the live page for receipts. Re-run on a new commit to refresh.