This article documents a production-grade architecture for generating research-grounded therapeutic content. The system prioritizes verifiable artifacts (papers → structured extracts → scored outputs → claim cards) over unstructured text.
You can treat this as a “trust pipeline”: retrieve → normalize → extract → score → repair → persist → generate.
Large language models feel like software, but they don’t behave like software.
With conventional programs, behavior is mostly deterministic: if tests pass, you ship, and nothing changes until you change the code. With LLM systems, behavior can drift without touching a line—model updates, prompt edits, temperature changes, tool availability, retrieval results, context truncation, and shifts in real-world inputs all move the output distribution.
So “it seems to work” isn’t a strategy. Evals are how you turn an LLM feature from a demo into an engineered system you can:
This doc builds evals from first principles and anchors everything in a concrete example: a workflow that classifies job postings as Remote EU (or not), outputs a structured JSON contract, and attaches multiple scorers (deterministic + LLM-as-judge) to generate reliable evaluation signals.
An eval is a function:
Eval(input, output, context?, ground_truth?) → score + reason + metadata
A single scalar score is rarely enough. You want:
When you do this consistently, evals become the LLM equivalent of:
In LLM systems, quality is a vector.
Even if the final label is right, the output can still be unacceptable if:
So you don’t build one eval. You build a panel of scorers that measure different axes.
No model involved. Examples:
reason non-empty)Strengths: fast, repeatable, low variance Limitations: shallow; can’t grade nuance like “is this reason actually supported?”
Use a second model (the judge) to grade output against a rubric and evidence.
Strengths: can evaluate nuanced properties like grounding, rubric adherence, and relevance Limitations: cost/latency, judge variance, judge drift, and susceptibility to prompt hacking if unconstrained
In production, the winning pattern is: deterministic guardrails + rubric-based judge scoring + sampling.
Input:
titlelocationdescriptionOutput contract:
{
"isRemoteEU": true,
"confidence": "high",
"reason": "Short evidence-based justification."
}
Job posts are full of ambiguous and misleading phrases:
This creates exactly the kind of environment where “vibes” fail and measurement matters.
A practical production architecture separates:
If your most expensive scorers run inline on every request, your feature inherits their cost and latency. A workflow-first approach gives you options:
If you can’t state what “correct” means, you can’t measure it consistently.
Your rubric should define:
Structured output makes evaluation composable:
isRemoteEU separately from reason,A robust suite typically includes:
hasReason)Generic “relevance” is not enough. You need:
“Is
isRemoteEUcorrect under this rubric for this job text?”
A strong judge returns structured, actionable feedback:
score ∈ [0, 1]isCorrect booleanmainIssues[] (typed failure modes)reasoning (short justification)evidenceQuotes[] (snippets that support the judgment)The most important instruction to judges:
Use ONLY the job text + rubric. Do not infer missing facts.
Without this, your judge will “helpfully” hallucinate implied constraints, and your metric becomes untrustworthy.
Even with a schema, add simple checks:
reason.trim().length > 0reason (prevents rambling)These are cheap, stable, and catch silent regressions early.
In many real products, the worst failure is not “wrong label.” It’s unsupported justification.
A model can guess the right label but invent a reason. Users trust the reason more than the label. When the reason lies, trust is gone.
Useful grounding dimensions:
If a scorer returns “lower is better” (hallucination/toxicity), invert it so higher is always better:
This prevents endless mistakes in dashboards and thresholds.
You typically want three layers:
Examples:
A weighted score helps compare versions, but should not hide hard failures.
Store mainIssues[] and judge reasons so you can cluster and fix.
A single global average is rarely useful. You want to slice by meaningful features:
For Remote EU:
This turns “accuracy dropped” into “accuracy dropped specifically on CET-only job posts.”
A rubric becomes much easier to debug when you can visualize precedence rules.
Here’s an example decision tree (adapt to your policy):
This makes edge cases explicit and makes judge behavior easier to audit.
A practical scoring policy:
This gives you statistical visibility with bounded cost.
If you want deeper rigor:
If you output confidence: high|medium|low, treat it as a measurable claim.
Track:
A healthy confidence signal produces a separation like:
If “high” is only marginally better than “medium,” you’re emitting vibes, not confidence.
Evals are not a report card. They’re a loop.
mainIssues[]The key operational shift: you stop debating anecdotes and start shipping changes backed by measured deltas.
A golden set is a curated collection of test cases representing:
Even 50–200 examples catch a shocking amount of regression.
For Remote EU, include cases mentioning:
Run the golden set:
Judge scoring is powerful, but you must treat the judge prompt like production code.
If you run multiple judges or compare judge vs deterministic heuristics, disagreement highlights ambiguous cases worth:
Not every system should block on evals, but you can safely gate high-risk cases.
Common gates:
Once your agent calls tools (search, databases, parsers, RAG), evals expand to include:
This is where agentic systems often succeed or fail in production.
Without evals, you can demo. With evals, you can ship—and keep shipping.
A workflow-first pattern—rubric + schema + domain correctness judge + grounding diagnostics + sampling + feedback loop—turns an LLM from a “text generator” into an engineered system you can measure, monitor, and improve like any serious production service.
Remote job postings are noisy, inconsistent, and often misleading. A role is labeled “Remote”, but the actual constraints show up in one sentence buried halfway down the description:
This article breaks down a LangGraph System that pre-screens job postings using DeepSeek structured extraction, then applies deterministic rules to instantly decide:
✅ Apply
❌ Reject (with reasons + quotes)
The goal is simple: filter out non-viable jobs before you spend time applying.
Keyword filters (“remote”, “anywhere”) fail because job descriptions are written inconsistently and constraints can be phrased in dozens of ways:
Instead of relying on fragile string matching, we use an LLM to read the description like a human, but output machine-usable constraints.
This agent evaluates job postings in two phases:
deepseek-chat)Instead of a linear script, the system is a graph-driven workflow:
This shape is production-friendly because the workflow can expand safely:
LangGraph becomes far more reliable when state is explicit:
class JobScreeningState(TypedDict):
job_title: str
company: str
description: str
location: str
url: str
# Candidate requirements
candidate_needs_visa_sponsorship: bool
requires_fully_remote: bool
requires_worldwide_remote: bool
candidate_locations: List[str]
# Output
is_eligible: bool
rejection_reasons: List[str]
# Extracted requirements
location_requirement: Optional[str]
specific_regions: List[str]
excluded_regions: List[str]
visa_sponsorship_available: Optional[bool]
work_authorization_required: Optional[str]
remote_status: str
timezone_restrictions: List[str]
confidence: str
key_phrases: List[str]
analysis_explanation: Optional[str]
Freeform LLM output is fragile. A production system needs predictable extraction. This agent forces DeepSeek into a strict schema:
class JobAnalysisSchema(TypedDict):
location_requirement: Literal["US-only", "EU-only", "Global", "Specific-regions", "Unknown"]
specific_regions: List[str]
excluded_regions: List[str]
remote_status: Literal["fully-remote", "remote-with-restrictions", "hybrid", "on-site", "unknown"]
visa_sponsorship_available: Optional[bool]
work_authorization_required: Literal["US-only", "EU-only", "Any", "Unknown"]
timezone_restrictions: List[str]
confidence: Literal["high", "medium", "low"]
key_phrases: List[str]
explanation: str
With this contract, the agent can safely feed extracted requirements into deterministic logic.
Job descriptions are long. Constraints are usually short. To reduce tokens and improve extraction precision, the system trims input to keyword-adjacent lines:
KEYWORDS = (
"remote", "anywhere", "worldwide", "timezone", "sponsor", "visa",
"authorized", "work authorization", "must be located", "eligible to work",
"location", "region", "country", "overlap", "hours", "time zone"
)
def _keep_relevant(text: str, window: int = 2) -> str:
lines = text.splitlines()
keep = set()
for i, ln in enumerate(lines):
if any(k in ln.lower() for k in KEYWORDS):
for j in range(max(0, i - window), min(len(lines), i + window + 1)):
keep.add(j)
return "\n".join(lines[i] for i in sorted(keep)) or text
This improves the system in four ways:
Before invoking DeepSeek, the system runs a tiny heuristic pre-check:
def _fast_heuristic_precheck(state: JobScreeningState) -> Optional[Dict[str, Any]]:
loc = state.get("location", "") or ""
desc = state.get("description", "") or ""
seed: Dict[str, Any] = {}
if _looks_worldwide(loc) or _looks_worldwide(desc):
seed["location_requirement"] = "Global"
seed["remote_status"] = "fully-remote"
if (_looks_us_only(loc) or _looks_us_only(desc)) and not seed.get("location_requirement"):
seed["location_requirement"] = "US-only"
if _looks_hybrid_or_onsite(loc):
seed["remote_status"] = "hybrid"
return seed if seed else None
DeepSeek still performs the full extraction, but seeding improves resilience against incomplete metadata.
The most valuable mode is strict worldwide remote filtering:
If requires_worldwide_remote=True, the job must satisfy ALL of the following:
remote_status == "fully-remote"location_requirement == "Global"specific_regionstimezone_restrictionsif state["requires_worldwide_remote"]:
if state["remote_status"] != "fully-remote":
rejection_reasons.append(
f"Not worldwide-remote: remote status is '{state['remote_status']}'"
)
if state["location_requirement"] != "Global":
rejection_reasons.append(
f"Not worldwide-remote: location requirement is '{state['location_requirement']}'"
)
if state["specific_regions"]:
rejection_reasons.append(
f"Not worldwide-remote: restricted to {state['specific_regions']}"
)
if state["timezone_restrictions"]:
rejection_reasons.append(
f"Not worldwide-remote: timezone restrictions {state['timezone_restrictions']}"
)
This instantly rejects “remote marketing” jobs like:
Sponsorship logic is easy to get wrong. The correct behavior:
False)None)if state["candidate_needs_visa_sponsorship"]:
if state["visa_sponsorship_available"] is False:
rejection_reasons.append(
"Job does not offer visa sponsorship, but candidate needs sponsorship"
)
This avoids dropping jobs that simply don’t mention sponsorship.
Trust requires receipts. The system stores:
rejection_reasons (deterministic outcomes)key_phrases (quotes that triggered the decision)analysis_explanation (LLM summary for debugging)That produces outputs like:
The included test suite covers the most common job board traps:
This validates both extraction quality and deterministic enforcement.
Binary decisions are clean, but scoring improves ranking:
You already compute a stable thread_id from the job URL. Persist results keyed by:
This prevents re-analyzing duplicate postings.
Add signals for:
This is one of the highest ROI improvements for global applicants.
This LangGraph System turns job descriptions into enforceable constraints:
This is how you scale job hunting without wasting time: automate rejection early, apply only where it can actually work.
Generating long-form audio content—audiobooks spanning hours, educational courses, or extended podcasts—presents unique challenges: API rate limits, network failures, resource constraints, and the sheer duration of processing. This article explores a production-ready architecture for long-running TTS pipelines that can gracefully handle long-form generation tasks, resume after failures, and maintain state across distributed systems.
Built with LangGraph, the system orchestrates complex workflows involving AI content generation (DeepSeek), text-to-speech conversion (OpenAI TTS), and distributed storage (Cloudflare R2). The key innovation: PostgreSQL checkpointing enables resumable execution, making it possible to generate 5-30+ minute audio segments reliably, even when individual API calls or processing steps fail.
Traditional TTS approaches fail at scale:
Our Solution: Stateful Orchestration
thread_id for job recoveryThe pipeline orchestrates three main workflows:
deepseek-chat, 2500+ token outputs)gpt-4o-mini-tts, 4096 char limit per request)Why This Stack for Long-Running Jobs:
The system implements three distinct LangGraph workflows, each optimized for specific tasks.
The research pipeline generates structured research content using a focused LangGraph workflow.
Key Features:
The most sophisticated workflow, supporting both full generation and audio-only modes.
Conditional Routing Logic:
def should_skip_text_generation(state: TextState) -> str:
"""Route to text generation or skip to audio."""
if state.get("existing_content") and state["existing_content"].get("text"):
return "chunk_text" # Audio-only mode
return "generate_text" # Full generation
def should_generate_audio(state: TextState) -> str:
"""Route to audio generation or end."""
if state.get("generate_audio", True):
return "chunk_text"
return END # Text-only mode
A simplified pipeline for generic long-form narration.
Iterative Chunk Processing:
The system uses a recursive edge pattern for processing chunks:
g.add_conditional_edges(
"tts_one_chunk",
edge_should_continue,
{
"tts_one_chunk": "tts_one_chunk", # Loop back
"finalize": "finalize", # Exit loop
},
)
def edge_should_continue(state: JobState) -> str:
if state["chunk_index"] < len(state["chunks"]):
return "tts_one_chunk"
return "finalize"
LangGraph uses typed state dictionaries for type safety and IDE support:
class TextState(TypedDict):
# Input metadata
content_id: int
title: str
content_type: str
language: str
target_duration_minutes: int | None
# Generation data
research_items: list[dict]
existing_content: dict | None
generated_text: str | None
# TTS fields
voice: str
chunks: List[str]
segment_urls: List[str]
manifest_url: Optional[str]
audio_url: Optional[str]
# Control flow
generate_audio: bool
database_saved: bool
error: str | None
For long-running jobs, checkpointing is non-negotiable. Without it, a network glitch at minute 25 of a 30-minute generation means restarting from scratch.
How Checkpointing Works:
async def run_pipeline(state: TextState, thread_id: str):
db_url = os.getenv("DATABASE_URL")
async with AsyncPostgresSaver.from_conn_string(db_url) as checkpointer:
await checkpointer.setup() # Creates checkpoint tables
app = build_graph(checkpointer=checkpointer)
config = {"configurable": {"thread_id": thread_id}}
# LangGraph automatically saves state after each node execution
final_state = await app.ainvoke(state, config=config)
return final_state
What Gets Checkpointed:
Recovery Example:
# Job crashes after generating 8 of 12 TTS segments
# Resume with same thread_id:
final_state = await run_pipeline(initial_state, thread_id="job-12345")
# LangGraph:
# 1. Loads last checkpoint from Postgres
# 2. Sees 8 segments already uploaded to R2
# 3. Continues from segment 9
# 4. Completes remaining 4 segments
Production Benefits:
thread_id valuesFor 30-minute audio (4,500+ words), naive chunking creates jarring transitions. Our algorithm balances API constraints with narrative flow:
Constraints:
Strategy:
def chunk_text(text: str, max_chars: int = 4000) -> List[str]:
"""
Multi-level chunking for long-form content:
1. Split by paragraphs (\n\n) - natural topic boundaries
2. Accumulate paragraphs until approaching 4K limit
3. If single paragraph > 4K, split by sentences
4. If single sentence > 4K, split mid-sentence (rare edge case)
Result: 10-15 chunks for 30-min audio, each ending at natural pause
"""
paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
chunks = []
buf = []
for p in paragraphs:
candidate = "\n\n".join(buf + [p]) if buf else p
if len(candidate) <= max_chars:
buf.append(p)
else:
if not buf:
# Paragraph too large - split by sentences
sentences = re.split(r"(?<=[.!?])\s+", p)
# Accumulate sentences with same logic...
else:
chunks.append("\n\n".join(buf))
buf = [p]
return chunks
Why This Matters for Long-Form:
Efficient audio generation using streaming responses:
async def node_tts_one_chunk(state: JobState) -> JobState:
chunk_text = state["chunks"][state["chunk_index"]]
segment_path = f"segment_{state['chunk_index']:04d}.mp3"
client = OpenAI()
# Stream directly to disk (memory efficient)
with client.audio.speech.with_streaming_response.create(
model="gpt-4o-mini-tts",
voice=state["voice"],
input=chunk_text,
response_format="mp3",
) as response:
response.stream_to_file(segment_path)
# Upload to R2
r2_url = upload_to_r2(segment_path, state["job_id"])
return {
**state,
"segment_urls": [*state["segment_urls"], r2_url],
"chunk_index": state["chunk_index"] + 1,
}
The system uses ffmpeg for high-quality concatenation:
async def node_generate_audio(state: TextState) -> TextState:
# Generate all segments...
# Create concat list for ffmpeg
file_list_path.write_text(
"\n".join(f"file '{segment}'" for segment in segment_paths)
)
# Merge using ffmpeg (codec copy - fast and lossless)
subprocess.run([
"ffmpeg", "-f", "concat", "-safe", "0",
"-i", str(file_list_path),
"-c", "copy", # No re-encoding
str(merged_path)
])
# Fallback to binary concatenation if ffmpeg unavailable
if not merged_path.exists():
with open(merged_path, "wb") as merged:
for segment in segment_paths:
merged.write(segment.read_bytes())
S3-compatible storage for globally distributed audio:
def get_r2_client():
return boto3.client(
's3',
endpoint_url=f'https://{R2_ACCOUNT_ID}.r2.cloudflarestorage.com',
aws_access_key_id=R2_ACCESS_KEY_ID,
aws_secret_access_key=R2_SECRET_ACCESS_KEY,
config=Config(signature_version='s3v4'),
)
def upload_to_r2(file_path: Path, job_id: str) -> str:
key = f"{job_id}/{file_path.name}"
client.put_object(
Bucket=R2_BUCKET_NAME,
Key=key,
Body=file_path.read_bytes(),
ContentType='audio/mpeg',
)
return f"{R2_PUBLIC_DOMAIN}/{key}"
The system implements a flexible content framework with customizable sections:
Key Components:
def build_content_prompt(state: TextState) -> str:
minutes = state.get("target_duration_minutes") or 5
target_words = int(minutes * 150) # 150 words per minute narration
content_type = state.get("content_type")
# Select architecture based on content type
architecture = generate_content_architecture(content_type)
return f"""
Create a {state['language']} narrative for audio:
TOPIC: {state['title']}
TYPE: {content_type}
TARGET: {target_words} words ({minutes} minutes)
{architecture}
RESEARCH CONTEXT:
{format_research_summary(state['research_items'])}
Requirements:
- Plain text only (no markdown)
- Natural paragraph breaks
- Engaging, clear tone
- Appropriate language for audio listening
"""
Endpoint Implementations:
@app.post("/api/research/generate")
async def research_endpoint(req: ResearchRequest):
"""Generate research context using LangGraph + DeepSeek."""
return await generate_research(req)
@app.post("/api/text/generate")
async def text_endpoint(req: TextGenerationRequest):
"""Generate long-form text content (text-only mode)."""
return await generate_text(req)
@app.post("/api/audio/generate")
async def audio_endpoint(req: TextGenerationRequest):
"""Generate audio from existing content (audio-only mode)."""
return await generate_audio(req)
@app.post("/api/tts/generate")
async def tts_endpoint(req: TTSRequest, background_tasks: BackgroundTasks):
"""Generic TTS generation (fire-and-forget)."""
return await generate_tts(req, background_tasks)
FROM python:3.12-slim
WORKDIR /app
# Install ffmpeg for audio merging
RUN apt-get update && apt-get install -y ffmpeg && rm -rf /var/lib/apt/lists/*
# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY . .
# Expose port
EXPOSE 8080
# Run FastAPI server
CMD ["uvicorn", "langgraph_server:app", "--host", "0.0.0.0", "--port", "8080"]
# AI Services
DEEPSEEK_API_KEY=sk-...
OPENAI_API_KEY=sk-...
# Database (Neon Postgres)
DATABASE_URL=postgresql://user:pass@host/db?sslmode=require
# Cloudflare R2
R2_ACCOUNT_ID=...
R2_ACCESS_KEY_ID=...
R2_SECRET_ACCESS_KEY=...
R2_BUCKET_NAME=longform-tts
R2_PUBLIC_DOMAIN=https://pub-longform-tts.r2.dev
# wrangler.toml
name = "langgraph-tts"
compatibility_date = "2024-01-01"
[build]
command = "docker build -t langgraph-tts ."
[[services]]
name = "langgraph-tts"
image = "langgraph-tts:latest"
[env]
PORT = "8080"
[[r2_buckets]]
binding = "LONGFORM_TTS"
bucket_name = "longform-tts"
Benchmarks (30-minute audio generation):
| Stage | Duration | Checkpointed | Retryable |
|---|---|---|---|
| Text Generation (DeepSeek) | 30-60s | ✅ After completion | ✅ Full retry |
| Text Chunking | <1s | ✅ After completion | ✅ Instant |
| TTS Segments (1–12) | 10-20s each | ✅ After each segment | ✅ Per-segment |
| Audio Merging (ffmpeg) | 1–3s | ✅ After completion | ✅ Full retry |
| R2 Upload (merged) | 2-5s | ✅ After completion | ✅ Full retry |
| Total Pipeline | 3-5 minutes | 15+ checkpoints | Granular recovery |
Long-Running Job Profile:
# Example: 2-hour audiobook chapter
text_length = 18,000 words
chunks = 45 # ~4,000 chars each
tts_time = 45 * 15s = 11.25 minutes
text_gen_time = 2-3 minutes
total_time = ~15 minutes for 2-hour audio
# Checkpoint frequency:
# - 1 after text generation
# - 45 after each TTS segment
# - 1 after merge
# Total: 47 recovery points
Failure Recovery Times:
async def node_generate_text(state: TextState) -> TextState:
try:
llm = ChatDeepSeek(model="deepseek-chat", temperature=0.7, max_tokens=2500)
prompt = build_therapeutic_prompt(state)
resp = await llm.ainvoke([HumanMessage(content=prompt)])
text = clean_for_tts(resp.content)
return {**state, "generated_text": text, "error": None}
except Exception as e:
print(f"❌ Text generation failed: {e}")
return {**state, "error": str(e)}
Key metrics to track:
Generation Metrics:
Quality Metrics:
Infrastructure Metrics:
Cost Metrics:
Horizontal Scaling:
Batch Processing:
async def batch_generate_audio(goal_ids: List[int]):
"""Process multiple goals in parallel."""
tasks = [run_pipeline(build_state(id), f"batch-{id}") for id in goal_ids]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
Queue-Based Processing:
# Optimize chunk size for TTS quality vs API limits
OPTIMAL_CHUNK_SIZE = 3000 # Sweet spot for natural pauses
# Parallel TTS generation (with rate limiting)
async def parallel_tts_generation(chunks: List[str], max_concurrent: int = 3):
semaphore = asyncio.Semaphore(max_concurrent)
async def generate_with_limit(chunk, index):
async with semaphore:
return await generate_tts_segment(chunk, index)
tasks = [generate_with_limit(c, i) for i, c in enumerate(chunks)]
return await asyncio.gather(*tasks)
# Cache research results for similar goals
@lru_cache(maxsize=100)
def get_research_for_goal_type(therapeutic_type: str, age: int):
"""Cache research by type + age bracket."""
return fetch_research(therapeutic_type, age)
# Cache text generation for re-use
async def get_or_generate_text(goal_id: int):
existing = await db.fetch_story(goal_id)
if existing and existing.created_at > datetime.now() - timedelta(days=7):
return existing.text
return await generate_new_text(goal_id)
def test_chunk_text_respects_limit():
long_text = "word " * 2000
chunks = chunk_text(long_text, max_chars=4000)
for chunk in chunks:
assert len(chunk) <= 4000
def test_clean_for_tts_removes_markdown():
text = "# Title\n\n**bold** and `code`"
cleaned = clean_for_tts(text)
assert "#" not in cleaned
assert "**" not in cleaned
assert "`" not in cleaned
@pytest.mark.asyncio
async def test_full_pipeline():
state = {
"goal_id": 1,
"goal_title": "Test anxiety reduction",
"therapeutic_goal_type": "anxiety_reduction",
"age": 8,
# ... other fields
}
result = await run_pipeline(state, "test-thread-1")
assert result["generated_text"] is not None
assert len(result["chunks"]) > 0
assert result["audio_url"] is not None
assert result["error"] is None
{workflow}-{entity_id}-{timestamp}# Support character dialogue with different voices
voices = {
"narrator": "cedar",
"child_character": "nova",
"parent_character": "marin"
}
# Adjust voice parameters based on content emotion
def get_tts_params(text: str) -> dict:
sentiment = analyze_sentiment(text)
if sentiment == "calm":
return {"speed": 0.9, "pitch": 0}
elif sentiment == "energetic":
return {"speed": 1.1, "pitch": 2}
# Stream audio as it's generated (SSE)
async def stream_audio_generation(goal_id: int):
async for chunk_url in generate_audio_stream(goal_id):
yield f"data: {json.dumps({'chunk_url': chunk_url})}\n\n"
This LangGraph-based TTS architecture demonstrates several key patterns:
The system successfully processes 5-30+ minute long-form narratives (up to 7,000+ words), generating research-backed content, converting to high-quality audio, and delivering via CDN—all while maintaining resumability after failures and full observability.
Real-World Performance:
Key Takeaways for Long-Running Pipelines:
The architecture scales to long-form audio generation: audiobooks (10+ hours), comprehensive courses, documentary narration, or serialized storytelling—any use case where reliability and resumability are non-negotiable.
This architecture powers long-form audio generation, combining LangGraph orchestration, OpenAI TTS streaming, and distributed storage for production-ready AI audio systems.
This guide demonstrates real-world implementation of DeFiTuna limit orders on Solana mainnet, focusing on:
Live Mainnet Transactions:
DeFiTuna combines Orca Whirlpools with on-chain limit orders, enabling automated trading triggers without requiring active monitoring. Unlike traditional AMMs where you passively provide liquidity, DeFiTuna positions execute predefined trades when price conditions are met.
This article walks through the complete implementation—from deriving PDAs to encoding instruction data—using Rust with solana-sdk and insights from the DeFiTuna SDK.
DeFiTuna doesn't implement its own AMM. Instead, it:
From our mainnet position 8wvKhHXHfzY4eQZTyK4kTfUtGj46XX5UX8P4S5kBbJ5:
Offset Field Type Bytes Description
------ ----- ---- ----- -----------
0-8 Discriminator u64 8 Account type identifier
8-40 Authority Pubkey 32 Position owner
40-72 Pool Pubkey 32 Orca Whirlpool address
72-73 Position Token u8 1 0=TokenA, 1=TokenB
73-74 Collateral Token u8 1 0=TokenA, 1=TokenB
92-100 Amount u64 8 Position size
100-108 Borrowed u64 8 Borrowed amount
184-200 Lower Limit √Price u128 16 Buy trigger price
200-216 Upper Limit √Price u128 16 Sell trigger price
Hex dump from mainnet:
00b8: 60 e4 c0 d6 1c 8e 68 01 00 00 00 00 00 00 00 00 Lower limit
00c8: 70 17 34 50 e5 c9 7c 01 00 00 00 00 00 00 00 00 Upper limit
These bytes encode:
6651068162312125808640 → $1307024310870365581606912 → $145bots/defituna-bot/
├── Cargo.toml
├── .env # Configuration
├── defituna.json # Program IDL
└── src/
├── config.rs # Shared config
└── bin/
├── open_spot_position.rs # Create position
├── set_limit_orders.rs # Configure triggers
├── check_position.rs # Query state
└── close_position.rs # Cleanup
File: src/bin/open_spot_position.rs
use solana_sdk::{
instruction::{AccountMeta, Instruction},
signature::{Keypair, Signer},
transaction::Transaction,
};
use spl_associated_token_account::get_associated_token_address;
// DeFiTuna program ID (mainnet)
const DEFITUNA_PROGRAM: &str = "tuna4uSQZncNeeiAMKbstuxA9CUkHH6HmC64wgmnogD";
// Orca SOL/USDC Whirlpool
const WHIRLPOOL: &str = "Czfq3xZZDmsdGdUyrNLtRhGc47cXcZtLG4crryfu44zE";
fn main() -> Result<()> {
let program_id = Pubkey::from_str(DEFITUNA_PROGRAM)?;
let whirlpool = Pubkey::from_str(WHIRLPOOL)?;
let authority = executor_keypair.pubkey();
// Derive spot position PDA
let (tuna_spot_position, _bump) = Pubkey::find_program_address(
&[
b"tuna_spot_position",
authority.as_ref(),
whirlpool.as_ref(),
],
&program_id,
);
// Token program IDs
let token_program_a = spl_token::ID; // SOL uses standard program
let token_program_b = spl_token::ID; // USDC uses standard program
// Derive associated token accounts for position
let tuna_position_ata_a = get_associated_token_address(
&tuna_spot_position,
&SOL_MINT
);
let tuna_position_ata_b = get_associated_token_address(
&tuna_spot_position,
&USDC_MINT
);
// Build instruction
// Discriminator from IDL: [87, 208, 173, 48, 231, 62, 210, 220]
let mut data = vec![87, 208, 173, 48, 231, 62, 210, 220];
// Args: position_token (PoolToken::A=0), collateral_token (PoolToken::B=1)
data.push(0); // Trading SOL (position_token = A)
data.push(1); // Collateral is USDC (collateral_token = B)
let instruction = Instruction {
program_id,
accounts: vec![
AccountMeta::new(authority, true), // authority (signer, writable)
AccountMeta::new_readonly(SOL_MINT, false), // mint_a
AccountMeta::new_readonly(USDC_MINT, false), // mint_b
AccountMeta::new_readonly(token_program_a, false), // token_program_a
AccountMeta::new_readonly(token_program_b, false), // token_program_b
AccountMeta::new(tuna_spot_position, false), // tuna_position (writable)
AccountMeta::new(tuna_position_ata_a, false), // tuna_position_ata_a
AccountMeta::new(tuna_position_ata_b, false), // tuna_position_ata_b
AccountMeta::new_readonly(whirlpool, false), // pool (Orca Whirlpool)
AccountMeta::new_readonly(system_program::ID, false), // system_program
AccountMeta::new_readonly(
spl_associated_token_account::ID,
false
), // associated_token_program
],
data,
};
// Create and sign transaction
let recent_blockhash = rpc_client.get_latest_blockhash()?;
let transaction = Transaction::new_signed_with_payer(
&[instruction],
Some(&authority),
&[&executor_keypair],
recent_blockhash,
);
// Send to RPC
let signature = rpc_client.send_and_confirm_transaction(&transaction)?;
println!("Position created: {}", signature);
Ok(())
}
Key Points:
["tuna_spot_position", authority, whirlpool]File: src/bin/set_limit_orders.rs
/// Convert price to sqrt_price format
/// Formula: sqrt_price = sqrt(price) * 2^64, adjusted for decimals
fn price_to_sqrt_price(price: f64, decimals_a: u8, decimals_b: u8) -> u128 {
let decimal_diff = decimals_a as i32 - decimals_b as i32;
let adjusted_price = if decimal_diff >= 0 {
price * 10_f64.powi(decimal_diff)
} else {
price / 10_f64.powi(-decimal_diff)
};
let sqrt_price_f64 = adjusted_price.sqrt() * (1u128 << 64) as f64;
sqrt_price_f64 as u128
}
fn main() -> Result<()> {
// Derive same position PDA
let (tuna_spot_position, _) = Pubkey::find_program_address(
&[b"tuna_spot_position", authority.as_ref(), whirlpool.as_ref()],
&program_id,
);
// Set limit prices
let lower_price = 130.0; // Buy if SOL drops to $130
let upper_price = 145.0; // Sell if SOL rises to $145
// Convert to sqrt_price (SOL=9 decimals, USDC=6 decimals)
let lower_sqrt_price = price_to_sqrt_price(lower_price, 9, 6);
let upper_sqrt_price = price_to_sqrt_price(upper_price, 9, 6);
// Build instruction
// Discriminator: [10, 180, 19, 205, 169, 133, 52, 118]
let mut data = vec![10, 180, 19, 205, 169, 133, 52, 118];
// Args: lower_limit_order_sqrt_price (u128), upper_limit_order_sqrt_price (u128)
data.extend_from_slice(&lower_sqrt_price.to_le_bytes());
data.extend_from_slice(&upper_sqrt_price.to_le_bytes());
let instruction = Instruction {
program_id,
accounts: vec![
AccountMeta::new_readonly(authority, true), // authority (signer)
AccountMeta::new(tuna_spot_position, false), // tuna_position (writable)
],
data,
};
let transaction = Transaction::new_signed_with_payer(
&[instruction],
Some(&authority),
&[&executor_keypair],
recent_blockhash,
);
let signature = rpc_client.send_and_confirm_transaction(&transaction)?;
println!("Limit orders set: {}", signature);
Ok(())
}
Output (mainnet):
Lower (buy): $130 → sqrt_price: 6651068162312125808640
Upper (sell): $145 → sqrt_price: 7024310870365581606912
Verification on-chain:
solana account 8wvKhHXHfzY4eQZTyK4kTfUtGj46XX5UX8P4S5kBbJ5
# Bytes 184-200 (lower limit):
60 e4 c0 d6 1c 8e 68 01 00 00 00 00 00 00 00 00
# Bytes 200-216 (upper limit):
70 17 34 50 e5 c9 7c 01 00 00 00 00 00 00 00 00
File: src/bin/close_position.rs
fn main() -> Result<()> {
// Derive position PDA (same as open)
let (tuna_spot_position, _) = Pubkey::find_program_address(
&[b"tuna_spot_position", authority.as_ref(), whirlpool.as_ref()],
&program_id,
);
// Build close instruction
// Discriminator: [4, 189, 171, 84, 110, 220, 10, 8]
let data = vec![4, 189, 171, 84, 110, 220, 10, 8];
let instruction = Instruction {
program_id,
accounts: vec![
AccountMeta::new(authority, true), // authority
AccountMeta::new_readonly(SOL_MINT, false), // mint_a
AccountMeta::new_readonly(USDC_MINT, false), // mint_b
AccountMeta::new_readonly(spl_token::ID, false), // token_program_a
AccountMeta::new_readonly(spl_token::ID, false), // token_program_b
AccountMeta::new(tuna_spot_position, false), // tuna_position
AccountMeta::new(tuna_position_ata_a, false), // tuna_position_ata_a
AccountMeta::new(tuna_position_ata_b, false), // tuna_position_ata_b
],
data,
};
let signature = rpc_client.send_and_confirm_transaction(&transaction)?;
println!("Position closed, rent recovered: {}", signature);
Ok(())
}
Result: Rent (0.00329904 SOL) returned to wallet, position account deleted.
From DefiTuna/tuna-sdk repository:
// ts-sdk/client/src/txbuilder/openTunaSpotPosition.ts
export async function openTunaSpotPositionInstructions(
rpc: Rpc<GetAccountInfoApi & GetMultipleAccountsApi>,
authority: TransactionSigner,
poolAddress: Address,
args: OpenTunaSpotPositionInstructionDataArgs,
): Promise<IInstruction[]> {
// 1. Derive position PDA
const tunaPositionAddress = (
await getTunaSpotPositionAddress(authority.address, poolAddress)
)[0];
// 2. Get associated token accounts
const tunaPositionAtaA = (
await findAssociatedTokenPda({
owner: tunaPositionAddress,
mint: mintA.address,
tokenProgram: mintA.programAddress,
})
)[0];
const tunaPositionAtaB = (
await findAssociatedTokenPda({
owner: tunaPositionAddress,
mint: mintB.address,
tokenProgram: mintB.programAddress,
})
)[0];
// 3. Build instruction
return getOpenTunaSpotPositionInstruction({
authority,
mintA: mintA.address,
mintB: mintB.address,
tokenProgramA: mintA.programAddress,
tokenProgramB: mintB.programAddress,
tunaPosition: tunaPositionAddress,
tunaPositionAtaA,
tunaPositionAtaB,
pool: poolAddress,
associatedTokenProgram: ASSOCIATED_TOKEN_PROGRAM_ADDRESS,
...args,
});
}
From rust-sdk/client/src/txbuilder/open_tuna_spot_position.rs:
pub fn open_tuna_spot_position_instructions(
rpc: &RpcClient,
authority: &Pubkey,
pool_address: &Pubkey,
args: OpenTunaSpotPositionInstructionArgs,
) -> Result<Vec<Instruction>> {
// 1. Fetch pool account to determine token mints
let pool_account = rpc.get_account(pool_address)?;
// Decode based on program owner
let (mint_a_address, mint_b_address) = if pool_account.owner == FUSIONAMM_ID {
let pool: FusionPool = decode_account(&pool_account)?;
(pool.token_mint_a, pool.token_mint_b)
} else if pool_account.owner == WHIRLPOOL_ID {
let pool: Whirlpool = decode_account(&pool_account)?;
(pool.token_mint_a, pool.token_mint_b)
} else {
return Err(anyhow!("Unsupported pool type"));
};
// 2. Get mint accounts to determine token programs
let mint_accounts = rpc.get_multiple_accounts(&[
mint_a_address.into(),
mint_b_address.into()
])?;
let mint_a_account = mint_accounts[0].as_ref()
.ok_or(anyhow!("Token A mint not found"))?;
let mint_b_account = mint_accounts[1].as_ref()
.ok_or(anyhow!("Token B mint not found"))?;
// 3. Build instruction
Ok(vec![open_tuna_spot_position_instruction(
authority,
pool_address,
&mint_a_address,
&mint_b_address,
&mint_a_account.owner, // Token program
&mint_b_account.owner,
args,
)])
}
Key SDK Features:
use solana_client::rpc_client::RpcClient;
use solana_sdk::commitment_config::CommitmentConfig;
let rpc_client = RpcClient::new_with_commitment(
"https://api.mainnet-beta.solana.com",
CommitmentConfig::confirmed()
);
// Fetch account with specific commitment
let account = rpc_client.get_account_with_commitment(
&position_address,
CommitmentConfig::finalized()
)?;
if let Some(account) = account.value {
println!("Account exists: {} bytes", account.data.len());
}
// Build transaction
let transaction = Transaction::new_signed_with_payer(
&[instruction],
Some(&payer.pubkey()),
&[&payer],
recent_blockhash,
);
// Simulate first
match rpc_client.simulate_transaction(&transaction) {
Ok(result) => {
if let Some(err) = result.value.err {
println!("Simulation failed: {:?}", err);
if let Some(logs) = result.value.logs {
for log in logs {
println!(" {}", log);
}
}
return Err(anyhow!("Simulation error"));
}
println!("✅ Simulation successful");
}
Err(e) => {
println!("RPC error during simulation: {}", e);
return Err(e.into());
}
}
// Now send for real
let signature = rpc_client.send_and_confirm_transaction(&transaction)?;
fn parse_position_limits(account_data: &[u8]) -> Result<(u128, u128)> {
if account_data.len() < 216 {
return Err(anyhow!("Account data too short"));
}
// Extract limit order sqrt prices
let lower_bytes: [u8; 16] = account_data[184..200]
.try_into()
.unwrap();
let upper_bytes: [u8; 16] = account_data[200..216]
.try_into()
.unwrap();
let lower_sqrt_price = u128::from_le_bytes(lower_bytes);
let upper_sqrt_price = u128::from_le_bytes(upper_bytes);
Ok((lower_sqrt_price, upper_sqrt_price))
}
// Usage
let account = rpc_client.get_account(&position_pda)?;
let (lower, upper) = parse_position_limits(&account.data)?;
// Convert to human-readable prices
let lower_price = sqrt_price_to_price(lower, 9, 6);
let upper_price = sqrt_price_to_price(upper, 9, 6);
println!("Buy limit: ${:.2}", lower_price);
println!("Sell limit: ${:.2}", upper_price);
use solana_sdk::signature::Signature;
use std::time::Duration;
fn wait_for_confirmation(
rpc_client: &RpcClient,
signature: &Signature,
max_retries: u32,
) -> Result<()> {
for i in 0..max_retries {
std::thread::sleep(Duration::from_secs(2));
if let Ok(status) = rpc_client.get_signature_status(signature) {
if let Some(result) = status {
if let Err(e) = result {
return Err(anyhow!("Transaction failed: {:?}", e));
}
println!("✅ Transaction confirmed in slot");
return Ok(());
}
}
if i == max_retries - 1 {
return Err(anyhow!("Transaction timeout"));
}
}
Ok(())
}
The modify_tuna_spot_position_orca instruction adds collateral and activates trading:
// From IDL
pub struct ModifyTunaSpotPositionOrca {
pub decrease_percent: u32, // 0 for increase
pub collateral_amount: u64, // USDC to deposit
pub borrow_amount: u64, // SOL to borrow from vault
pub required_swap_amount: u64, // 0 for auto-calc
pub remaining_accounts_info: RemainingAccountsInfo,
}
// Required accounts (24 total)
accounts: vec![
AccountMeta::new(authority, true),
AccountMeta::new_readonly(tuna_config, false),
AccountMeta::new_readonly(mint_a, false),
AccountMeta::new_readonly(mint_b, false),
AccountMeta::new_readonly(token_program_a, false),
AccountMeta::new_readonly(token_program_b, false),
AccountMeta::new(market, false),
AccountMeta::new(vault_a, false),
AccountMeta::new(vault_b, false),
AccountMeta::new(vault_a_ata, false),
AccountMeta::new(vault_b_ata, false),
AccountMeta::new(tuna_spot_position, false),
AccountMeta::new(tuna_position_ata_a, false),
AccountMeta::new(tuna_position_ata_b, false),
AccountMeta::new(authority_ata_a, false),
AccountMeta::new(authority_ata_b, false),
AccountMeta::new(fee_recipient_ata_a, false),
AccountMeta::new(fee_recipient_ata_b, false),
AccountMeta::new_readonly(pyth_oracle_price_feed_a, false),
AccountMeta::new_readonly(pyth_oracle_price_feed_b, false),
AccountMeta::new_readonly(whirlpool_program, false),
AccountMeta::new(whirlpool, false),
AccountMeta::new_readonly(memo_program, false),
AccountMeta::new_readonly(system_program::ID, false),
]
Why so many accounts?
DeFiTuna stores prices as (same as Uniswap V3):
Example: SOL/USDC at $130
fn sqrt_price_to_price(sqrt_price: u128, decimals_a: u8, decimals_b: u8) -> f64 {
let decimal_diff = decimals_a as i32 - decimals_b as i32;
let price_raw = (sqrt_price as f64 / (1u128 << 64) as f64).powi(2);
if decimal_diff >= 0 {
price_raw / 10_f64.powi(decimal_diff)
} else {
price_raw * 10_f64.powi(-decimal_diff)
}
}
// Verify on-chain data
let lower_sqrt_price = 6651068162312125808640u128;
let price = sqrt_price_to_price(lower_sqrt_price, 9, 6);
assert_eq!(price, 130.0);
# DeFiTuna pools don't exist on devnet
$ solana account 9m96e4CieVMjTC7vP1a1pM3qfn5A5kHRPs3SrsVZBGqt --url devnet
Error: AccountNotFound
# Orca Whirlpools also limited on devnet
Recommendation: Test on mainnet with minimal amounts (0.01-0.1 SOL).
Total cost: ~$0.002 for full testing cycle
#[derive(Debug)]
pub enum DefiTunaError {
#[error("Position does not exist")]
PositionNotFound,
#[error("Invalid tick range")]
InvalidTickRange,
#[error("Insufficient collateral")]
InsufficientCollateral,
#[error("RPC error: {0}")]
RpcError(#[from] solana_client::client_error::ClientError),
}
// Usage
match rpc_client.get_account(&position_pda) {
Ok(_) => { /* process */ },
Err(_) => return Err(DefiTunaError::PositionNotFound),
}
use std::time::{Duration, Instant};
struct RateLimiter {
last_request: Instant,
min_interval: Duration,
}
impl RateLimiter {
pub fn new(requests_per_second: u32) -> Self {
Self {
last_request: Instant::now(),
min_interval: Duration::from_millis(1000 / requests_per_second as u64),
}
}
pub fn wait(&mut self) {
let elapsed = self.last_request.elapsed();
if elapsed < self.min_interval {
std::thread::sleep(self.min_interval - elapsed);
}
self.last_request = Instant::now();
}
}
// Usage with public RPC (limit to 10 req/s)
let mut limiter = RateLimiter::new(10);
for position in positions {
limiter.wait();
let account = rpc_client.get_account(&position)?;
// process...
}
const MAX_RETRIES: u32 = 3;
fn send_with_retry(
rpc_client: &RpcClient,
transaction: &Transaction,
) -> Result<Signature> {
let mut last_error = None;
for attempt in 1..=MAX_RETRIES {
match rpc_client.send_and_confirm_transaction(transaction) {
Ok(signature) => return Ok(signature),
Err(e) => {
println!("Attempt {}/{} failed: {}", attempt, MAX_RETRIES, e);
last_error = Some(e);
if attempt < MAX_RETRIES {
std::thread::sleep(Duration::from_secs(2_u64.pow(attempt)));
}
}
}
}
Err(last_error.unwrap().into())
}
Here's a full program that opens a position, sets limits, and closes:
use anyhow::Result;
use solana_client::rpc_client::RpcClient;
use solana_sdk::{
instruction::{AccountMeta, Instruction},
pubkey::Pubkey,
signature::{Keypair, Signer},
transaction::Transaction,
};
use std::str::FromStr;
const DEFITUNA_PROGRAM: &str = "tuna4uSQZncNeeiAMKbstuxA9CUkHH6HmC64wgmnogD";
const WHIRLPOOL: &str = "Czfq3xZZDmsdGdUyrNLtRhGc47cXcZtLG4crryfu44zE";
const SOL_MINT: &str = "So11111111111111111111111111111111111111112";
const USDC_MINT: &str = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v";
fn main() -> Result<()> {
// Initialize
let rpc_url = "https://api.mainnet-beta.solana.com";
let rpc_client = RpcClient::new(rpc_url);
let executor_keypair = read_keypair_from_env()?;
let program_id = Pubkey::from_str(DEFITUNA_PROGRAM)?;
let whirlpool = Pubkey::from_str(WHIRLPOOL)?;
// Derive position PDA
let (position_pda, _) = Pubkey::find_program_address(
&[
b"tuna_spot_position",
executor_keypair.pubkey().as_ref(),
whirlpool.as_ref(),
],
&program_id,
);
println!("Position PDA: {}", position_pda);
// Step 1: Open position
let open_ix = build_open_position_instruction(
&program_id,
&executor_keypair.pubkey(),
&whirlpool,
&position_pda,
)?;
let sig1 = send_instruction(&rpc_client, &executor_keypair, open_ix)?;
println!("✅ Position opened: {}", sig1);
// Step 2: Set limit orders
let limits_ix = build_set_limits_instruction(
&program_id,
&executor_keypair.pubkey(),
&position_pda,
130.0, // buy at $130
145.0, // sell at $145
)?;
let sig2 = send_instruction(&rpc_client, &executor_keypair, limits_ix)?;
println!("✅ Limits set: {}", sig2);
// Step 3: Verify on-chain
std::thread::sleep(std::time::Duration::from_secs(5));
let account = rpc_client.get_account(&position_pda)?;
let (lower, upper) = parse_limits(&account.data)?;
println!("📊 On-chain limits: ${:.2} / ${:.2}", lower, upper);
// Step 4: Close position
let close_ix = build_close_position_instruction(
&program_id,
&executor_keypair.pubkey(),
&position_pda,
)?;
let sig3 = send_instruction(&rpc_client, &executor_keypair, close_ix)?;
println!("✅ Position closed: {}", sig3);
Ok(())
}
fn send_instruction(
rpc_client: &RpcClient,
payer: &Keypair,
instruction: Instruction,
) -> Result<String> {
let recent_blockhash = rpc_client.get_latest_blockhash()?;
let transaction = Transaction::new_signed_with_payer(
&[instruction],
Some(&payer.pubkey()),
&[payer],
recent_blockhash,
);
let signature = rpc_client.send_and_confirm_transaction(&transaction)?;
Ok(signature.to_string())
}
// Helper functions omitted for brevity
// See full implementation in GitHub repository
Before deploying to production:
Building on DeFiTuna requires understanding:
The mainnet transactions in this guide prove that limit orders are truly stored on-chain and executable without active monitoring. Once set, the DeFiTuna protocol monitors prices and executes trades automatically.
Key Insight: DeFiTuna is a protocol abstraction layer, not a standalone AMM. It wraps existing liquidity sources (Orca, Fusion) with advanced order types, making it a powerful tool for automated trading strategies on Solana.
TL;DR: Deploy a production-ready real-time Solana slot streaming service using Helius LaserStream SDK on Cloudflare Containers.
Helius LaserStream provides ultra-low latency access to Solana data via gRPC streaming. Traditional WebSocket polling introduces delays; LaserStream eliminates this with direct gRPC connections to Helius nodes.
Traditional deployments require server provisioning, load balancing, and scaling. Cloudflare Containers solve this:
| Feature | Traditional VPS | Cloudflare Containers |
|---|---|---|
| Global deployment | Manual multi-region setup | Automatic edge deployment |
| Scaling | Manual or autoscaling groups | Auto-scaling via Durable Objects |
| Cold start | Always running (cost) | Sleep after inactivity |
| gRPC support | Yes | Yes (in Containers, not Workers) |
| SSL/TLS | Manual cert management | Automatic |
| DDoS protection | Additional service | Built-in |
Key components:
laserstream-container/
├── src/
│ └── index.ts # Worker (Hono API + Durable Object routing)
├── container_src/
│ ├── Cargo.toml # Rust dependencies
│ └── src/
│ ├── main.rs # Axum HTTP server
│ └── stream.rs # LaserStream gRPC client
├── Dockerfile # Multi-stage Rust build
├── wrangler.jsonc # Cloudflare configuration
├── package.json # Build and deployment scripts
└── tsconfig.json # TypeScript configuration
Before deploying, ensure you have:
npm install -g wrangler)wrangler login
This opens a browser to authorize Wrangler with your Cloudflare account.
The Worker provides the HTTP API layer and routes requests to the container.
pnpm install @cloudflare/containers hono
pnpm install -D typescript wrangler
src/index.ts)import { Container } from "@cloudflare/containers";
import { Hono } from "hono";
// Define the container class
export class LaserStreamContainer extends Container<Env> {
defaultPort = 8080;
sleepAfter = "2m"; // Sleep after 2 minutes of inactivity
envVars = {
HELIUS_API_KEY: "", // Set via wrangler secret
LASERSTREAM_ENDPOINT: "https://laserstream-devnet-ewr.helius-rpc.com",
RUST_LOG: "info",
};
override onStart() {
console.log("LaserStream container started");
}
override onStop() {
console.log("LaserStream container stopped");
}
override onError(error: unknown) {
console.error("LaserStream container error:", error);
}
}
// Create Hono app
const app = new Hono<{ Bindings: Env }>();
// Service information endpoint
app.get("/", (c) => {
return c.text(
"LaserStream on Cloudflare Containers\n\n" +
"Endpoints:\n" +
"GET /health - Health check\n" +
"POST /start - Start LaserStream subscription\n" +
"GET /latest - Get latest slot update\n"
);
});
// Worker health check
app.get("/health", (c) => {
return c.json({
status: "ok",
timestamp: new Date().toISOString()
});
});
// Proxy all other requests to the singleton container
app.all("*", async (c) => {
try {
// Get singleton container by name
const containerId = c.env.LASERSTREAM_CONTAINER.idFromName("laserstream-main");
const container = c.env.LASERSTREAM_CONTAINER.get(containerId);
// Forward request to container
return await container.fetch(c.req.raw);
} catch (error) {
console.error("Container error:", error);
return c.json(
{ error: "Container unavailable", details: String(error) },
500
);
}
});
export default app;
idFromName("laserstream-main") ensures only one container instance handles all requestsenvVarsThe Rust container runs the LaserStream gRPC client and exposes an HTTP API.
container_src/Cargo.toml)[package]
name = "laserstream_container"
version = "0.1.0"
edition = "2021"
[dependencies]
anyhow = "1.0"
axum = "0.7"
chrono = { version = "0.4", features = ["serde"] }
futures-util = "0.3"
helius-laserstream = "0.1.5"
serde = { version = "1.0", features = ["derive"] }
tokio = { version = "1.49", features = ["full"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
container_src/src/main.rs)use std::{
net::SocketAddr,
sync::{
atomic::{AtomicBool, Ordering},
Arc,
},
};
use axum::{
extract::State,
http::StatusCode,
response::IntoResponse,
routing::{get, post},
Json, Router,
};
use serde::Serialize;
use tokio::sync::RwLock;
use tracing::{error, info};
use tracing_subscriber::EnvFilter;
mod stream;
#[derive(Clone)]
struct AppState {
started: Arc<AtomicBool>,
latest: Arc<RwLock<Option<LatestSlot>>>,
}
#[derive(Debug, Clone, Serialize)]
struct LatestSlot {
slot: u64,
parent: Option<u64>,
status: String,
created_at_rfc3339: Option<String>,
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
// Initialize logging
tracing_subscriber::fmt()
.with_env_filter(EnvFilter::from_default_env())
.init();
let port: u16 = std::env::var("PORT")
.unwrap_or_else(|_| "8080".to_string())
.parse()?;
let state = AppState {
started: Arc::new(AtomicBool::new(false)),
latest: Arc::new(RwLock::new(None)),
};
// Start stream on boot
ensure_stream_started(state.clone()).await;
// Define routes
let app = Router::new()
.route("/health", get(health))
.route("/start", post(start))
.route("/latest", get(latest))
.with_state(state);
let addr = SocketAddr::from(([0, 0, 0, 0], port));
info!("listening on {}", addr);
let listener = tokio::net::TcpListener::bind(addr).await?;
axum::serve(listener, app).await?;
Ok(())
}
async fn health() -> impl IntoResponse {
(StatusCode::OK, "ok\n")
}
async fn start(State(state): State<AppState>) -> impl IntoResponse {
ensure_stream_started(state).await;
(StatusCode::OK, "started\n")
}
async fn latest(State(state): State<AppState>) -> impl IntoResponse {
let guard = state.latest.read().await;
match guard.as_ref() {
Some(slot) => (StatusCode::OK, Json(slot.clone())),
None => (
StatusCode::NOT_FOUND,
Json(LatestSlot {
slot: 0,
parent: None,
status: "no data".to_string(),
created_at_rfc3339: None,
}),
),
}
}
async fn ensure_stream_started(state: AppState) {
if !state.started.swap(true, Ordering::SeqCst) {
tokio::spawn(async move {
if let Err(e) = stream::run_slot_stream(state.clone()).await {
error!("Stream error: {}", e);
state.started.store(false, Ordering::SeqCst);
}
});
}
}
container_src/src/stream.rs)use anyhow::{anyhow, Context};
use chrono::{DateTime, Utc};
use futures_util::StreamExt;
use tokio::pin;
use tracing::{info, warn};
use crate::{AppState, LatestSlot};
use helius_laserstream::{
config::LaserstreamConfig,
grpc::{subscribe_update::UpdateOneof, SubscribeRequest},
client::subscribe,
};
pub async fn run_slot_stream(state: AppState) -> anyhow::Result<()> {
let endpoint = std::env::var("LASERSTREAM_ENDPOINT")
.context("LASERSTREAM_ENDPOINT is required")?;
let api_key = std::env::var("HELIUS_API_KEY")
.context("HELIUS_API_KEY is required")?;
info!("connecting to LaserStream at {}", endpoint);
let config = LaserstreamConfig {
endpoint,
x_token: Some(api_key),
};
let request = SubscribeRequest {
slots: HashMap::new(),
accounts: HashMap::new(),
transactions: HashMap::new(),
blocks: HashMap::new(),
blocks_meta: HashMap::new(),
entry: HashMap::new(),
commitment: None,
accounts_data_slice: vec![],
ping: None,
};
let mut stream = subscribe(config, request).await?;
pin!(stream);
info!("LaserStream connected, waiting for updates...");
while let Some(msg) = stream.next().await {
match msg {
Ok(update) => {
if let Some(UpdateOneof::Slot(slot_update)) = update.update_oneof {
let created_at = slot_update
.created_at
.and_then(|ts| {
DateTime::from_timestamp(ts.seconds, ts.nanos as u32)
})
.map(|dt: DateTime<Utc>| dt.to_rfc3339());
let latest = LatestSlot {
slot: slot_update.slot,
parent: Some(slot_update.parent),
status: format!("{:?}", slot_update.status),
created_at_rfc3339: created_at,
};
*state.latest.write().await = Some(latest.clone());
info!("slot update: {:?}", latest);
}
}
Err(e) => {
warn!("stream error: {}", e);
return Err(anyhow!("stream error: {}", e));
}
}
}
Ok(())
}
Build the Rust container with a multi-stage Dockerfile for minimal image size.
# syntax=docker/dockerfile:1
FROM rust:1.83-slim AS build
WORKDIR /app
# Install build dependencies
RUN apt-get update && apt-get install -y \
pkg-config \
libssl-dev \
protobuf-compiler \
build-essential \
g++ \
&& rm -rf /var/lib/apt/lists/*
# Copy Rust source
COPY container_src/Cargo.toml ./
COPY container_src/src ./src
# Build release binary
RUN cargo build --release
# Runtime image
FROM debian:bookworm-slim
RUN apt-get update && \
apt-get install -y ca-certificates libssl3 && \
rm -rf /var/lib/apt/lists/*
COPY --from=build /app/target/release/laserstream_container /laserstream_container
EXPOSE 8080
CMD ["/laserstream_container"]
--releaseConfigure the Worker and Container deployment.
wrangler.jsonc{
"$schema": "node_modules/wrangler/config-schema.json",
"name": "laserstream-container",
"main": "src/index.ts",
"compatibility_date": "2025-01-08",
"compatibility_flags": ["nodejs_compat"],
"observability": {
"enabled": true
},
"containers": [
{
"class_name": "LaserStreamContainer",
"image": "registry.cloudflare.com/<ACCOUNT_ID>/laserstream-container-rust:v1.0.0",
"max_instances": 10
}
],
"durable_objects": {
"bindings": [
{
"class_name": "LaserStreamContainer",
"name": "LASERSTREAM_CONTAINER"
}
]
},
"migrations": [
{
"new_sqlite_classes": ["LaserStreamContainer"],
"tag": "v1"
}
],
"vars": {
"LASERSTREAM_ENDPOINT": "https://laserstream-devnet-ewr.helius-rpc.com"
}
}
Replace <ACCOUNT_ID> with your Cloudflare account ID (find it in the Cloudflare dashboard).
package.json){
"name": "laserstream-container",
"version": "1.0.0",
"scripts": {
"build": "tsc && cargo build --release --manifest-path=container_src/Cargo.toml",
"build:container": "wrangler containers build . --tag laserstream-container-rust:latest",
"push:container": "wrangler containers push laserstream-container-rust:latest",
"deploy": "wrangler deploy",
"dev": "wrangler dev",
"tail": "wrangler tail",
"secret:set": "wrangler secret put"
},
"dependencies": {
"@cloudflare/containers": "^0.0.21",
"hono": "4.11.1"
},
"devDependencies": {
"@types/node": "^25.0.3",
"typescript": "5.9.3",
"wrangler": "4.58.0"
}
}
# Build Rust container locally
pnpm run build:container
# Tag with version
docker tag laserstream-container-rust:latest laserstream-container-rust:v1.0.0
# Push to Cloudflare registry
pnpm run push:container
Expected output:
Building container image...
Successfully tagged laserstream-container-rust:latest
Pushing to registry.cloudflare.com/...
Image pushed successfully
Digest: sha256:59c03a69b057...
Before deploying, set the Helius API key:
echo "YOUR_HELIUS_API_KEY" | pnpm run secret:set HELIUS_API_KEY
Alternatively, use interactive mode:
pnpm run secret:set HELIUS_API_KEY
# Paste your API key when prompted
pnpm run deploy
Expected output:
Uploading Worker...
Published laserstream-container (0.42 sec)
https://laserstream-container.<your-subdomain>.workers.dev
curl https://laserstream-container.<your-subdomain>.workers.dev/health
Expected response:
{
"status": "ok",
"timestamp": "2025-01-09T12:34:56.789Z"
}
curl -X POST https://laserstream-container.<your-subdomain>.workers.dev/start
Expected response:
started
curl https://laserstream-container.<your-subdomain>.workers.dev/latest
Expected response:
{
"slot": 285432167,
"parent": 285432166,
"status": "Confirmed",
"created_at_rfc3339": "2025-01-09T12:35:01.234Z"
}
pnpm run tail
Expected output:
2025-01-09T12:34:56.789Z INFO laserstream_container: listening on 0.0.0.0:8080
2025-01-09T12:35:01.234Z INFO laserstream_container: connecting to LaserStream at https://laserstream-devnet-ewr.helius-rpc.com
2025-01-09T12:35:02.456Z INFO laserstream_container: LaserStream connected, waiting for updates...
2025-01-09T12:35:03.678Z INFO laserstream_container: slot update: LatestSlot { slot: 285432167, ... }
Cause: HELIUS_API_KEY secret not set or incorrect.
Fix:
# Verify secret is set
wrangler secret list
# Re-set if missing
echo "YOUR_KEY" | pnpm run secret:set HELIUS_API_KEY
# Redeploy
pnpm run deploy
Cause: Docker image not pushed or incorrect registry path.
Fix:
# Verify image exists
docker images | grep laserstream
# Rebuild and push
pnpm run build:container
pnpm run push:container
pnpm run deploy
Cause: Container sleeping or crashed.
Fix:
# Check logs
pnpm run tail
# Restart container
curl -X POST https://<your-url>/start
sleepAfter based on request frequencymax_instances based on expected loadexport class LaserStreamContainer extends Container<Env> {
sleepAfter = "5m"; // Sleep after 5 minutes for dev
// sleepAfter = "30m"; // Sleep after 30 minutes for production
}
Add retry logic and circuit breakers:
// In stream.rs
pub async fn run_slot_stream(state: AppState) -> anyhow::Result<()> {
let mut retry_count = 0;
const MAX_RETRIES: u32 = 5;
loop {
match try_connect(&state).await {
Ok(_) => {
retry_count = 0; // Reset on success
}
Err(e) => {
retry_count += 1;
if retry_count >= MAX_RETRIES {
return Err(anyhow!("Max retries exceeded: {}", e));
}
let backoff = std::time::Duration::from_secs(2_u64.pow(retry_count));
warn!("Retry {} after {:?}: {}", retry_count, backoff, e);
tokio::time::sleep(backoff).await;
}
}
}
}
For global low-latency access, use Cloudflare's automatic edge deployment:
{
"placement": { "mode": "smart" }
}
This automatically routes requests to the nearest Cloudflare edge location.
HELIUS_API_KEY// In Worker
app.use("*", async (c, next) => {
const token = c.req.header("Authorization");
if (!token || token !== `Bearer ${c.env.API_SECRET}`) {
return c.json({ error: "Unauthorized" }, 401);
}
await next();
});
// jupiter-laserstream-bot/src/poller.ts
import { setInterval } from "timers/promises";
const CONTAINER_URL = "https://laserstream-container.<subdomain>.workers.dev";
async function pollLatestSlot() {
const response = await fetch(`${CONTAINER_URL}/latest`);
const data = await response.json();
console.log(`Latest slot: ${data.slot}`);
// Trigger trading logic
await handleSlotUpdate(data);
}
// Poll every 2 seconds
for await (const _ of setInterval(2000)) {
await pollLatestSlot();
}
Convert HTTP polling to WebSocket for browser clients:
// websocket-bridge.ts
import { WebSocketServer } from "ws";
const wss = new WebSocketServer({ port: 8080 });
const CONTAINER_URL = "https://laserstream-container.<subdomain>.workers.dev";
wss.on("connection", (ws) => {
const interval = setInterval(async () => {
const response = await fetch(`${CONTAINER_URL}/latest`);
const data = await response.json();
ws.send(JSON.stringify(data));
}, 1000);
ws.on("close", () => clearInterval(interval));
});
| Approach | Latency | Cost | Complexity | Scalability |
|---|---|---|---|---|
| WebSocket polling | ~500ms | Low | Low | Manual |
| Traditional VPS | ~100ms | Medium | High | Manual |
| LaserStream + Cloudflare | ~50ms | Low (pay-per-use) | Medium | Automatic |
| Direct gRPC | ~30ms | Medium | High | Manual |
✅ Use Cloudflare Containers when:
❌ Use traditional VPS when:
Deploying LaserStream on Cloudflare Containers provides a production-ready solution for real-time Solana data streaming with:
The combination of Helius LaserStream's ultra-low latency gRPC streaming and Cloudflare's global network creates a powerful platform for building real-time Solana applications.
DefiTuna AMM combines concentrated liquidity, on-chain limit orders, and leveraged liquidity provision in a unique AMM design. This article focuses on the Anchor smart contract that implements this protocol, specifically examining:
initialize_pool, create_position, place_limit_order, swap, leverage_positionDefiTuna AMM represents an innovative approach to decentralized exchanges by integrating traditional AMM mechanics with orderbook-style limit orders and leverage capabilities. This Anchor smart contract implements the core on-chain logic that makes this possible, providing developers with a practical example of advanced Solana DeFi programming patterns.
The program follows a modular architecture with separate handlers for pool management, position creation, order placement, and swap execution. All state is managed through PDAs to ensure secure ownership and access control.
| Account | Type | PDA Seeds | Purpose |
|---|---|---|---|
Pool | State | ["pool", base_mint, quote_mint] | Stores pool configuration and global state |
Position | State | ["position", pool, owner, position_id] | Tracks user's concentrated liquidity position |
LimitOrder | State | ["order", pool, owner, order_id] | Stores limit order parameters and status |
LeveragedPosition | State | ["leverage", pool, owner, leverage_id] | Manages leveraged liquidity positions |
ProtocolConfig | State | ["config"] | Global protocol parameters and fee accounts |
Tick | State | ["tick", pool, tick_index] | Stores liquidity data for specific price ticks |
#[account]
#[derive(InitSpace)]
pub struct Pool {
pub base_mint: Pubkey,
pub quote_mint: Pubkey,
pub fee_rate: u16, // basis points
pub protocol_fee_rate: u16,
pub tick_spacing: i32,
pub current_tick_index: i32,
pub liquidity: u128,
pub sqrt_price: u128,
pub fee_growth_global_0: u128,
pub fee_growth_global_1: u128,
pub protocol_fees_0: u64,
pub protocol_fees_1: u64,
pub bump: u8,
}
// PDA derivation for pool
let (pool_pda, pool_bump) = Pubkey::find_program_address(
&[
b"pool",
base_mint.as_ref(),
quote_mint.as_ref(),
],
program_id
);
## Instruction Handlers Deep Dive
### 1. `initialize_pool` - Pool Creation
This instruction creates a new liquidity pool with specified parameters:
```rust
#[derive(Accounts)]
pub struct InitializePool<'info> {
#[account(mut)]
pub payer: Signer<'info>,
#[account(
init,
payer = payer,
space = 8 + Pool::INIT_SPACE,
seeds = [
b"pool",
base_mint.key().as_ref(),
quote_mint.key().as_ref()
],
bump
)]
pub pool: Account<'info, Pool>,
pub base_mint: Account<'info, Mint>,
pub quote_mint: Account<'info, Mint>,
pub system_program: Program<'info, System>,
}
pub fn initialize_pool(
ctx: Context<InitializePool>,
fee_rate: u16,
tick_spacing: i32,
initial_sqrt_price: u128,
) -> Result<()> {
let pool = &mut ctx.accounts.pool;
// Validate parameters
require!(fee_rate <= MAX_FEE_RATE, ErrorCode::InvalidFeeRate);
require!(tick_spacing > 0, ErrorCode::InvalidTickSpacing);
// Initialize pool state
pool.base_mint = ctx.accounts.base_mint.key();
pool.quote_mint = ctx.accounts.quote_mint.key();
pool.fee_rate = fee_rate;
pool.tick_spacing = tick_spacing;
pool.sqrt_price = initial_sqrt_price;
pool.current_tick_index = calculate_tick_from_sqrt_price(initial_sqrt_price);
pool.bump = ctx.bumps.pool;
Ok(())
}
### 2. `create_position` - Concentrated Liquidity Position
Creates a position with liquidity concentrated between specified ticks:
```rust
#[derive(Accounts)]
#[instruction(position_id: u64)]
pub struct CreatePosition<'info> {
#[account(mut)]
pub owner: Signer<'info>,
#[account(
seeds = [
b"pool",
pool.base_mint.as_ref(),
pool.quote_mint.as_ref()
],
bump = pool.bump
)]
pub pool: Account<'info, Pool>,
#[account(
init,
payer = owner,
space = 8 + Position::INIT_SPACE,
seeds = [
b"position",
pool.key().as_ref(),
owner.key().as_ref(),
&position_id.to_le_bytes()
],
bump
)]
pub position: Account<'info, Position>,
#[account(mut)]
pub token_account_a: Account<'info, TokenAccount>,
#[account(mut)]
pub token_account_b: Account<'info, TokenAccount>,
pub token_program: Program<'info, Token>,
pub system_program: Program<'info, System>,
}
#[account]
#[derive(InitSpace)]
pub struct Position {
pub pool: Pubkey,
pub owner: Pubkey,
pub liquidity: u128,
pub tick_lower: i32,
pub tick_upper: i32,
pub fee_growth_inside_0_last: u128,
pub fee_growth_inside_1_last: u128,
pub tokens_owed_0: u64,
pub tokens_owed_1: u64,
pub bump: u8,
}
pub fn create_position(
ctx: Context<CreatePosition>,
position_id: u64,
tick_lower: i32,
tick_upper: i32,
liquidity_delta: u128,
) -> Result<()> {
let pool = &mut ctx.accounts.pool;
let position = &mut ctx.accounts.position;
// Validate tick bounds
require!(tick_lower < tick_upper, ErrorCode::InvalidTickRange);
require!(tick_lower % pool.tick_spacing == 0, ErrorCode::TickNotSpaced);
require!(tick_upper % pool.tick_spacing == 0, ErrorCode::TickNotSpaced);
// Calculate required token amounts
let (amount_a, amount_b) = calculate_liquidity_amounts(
pool.sqrt_price,
tick_lower,
tick_upper,
liquidity_delta
);
// Transfer tokens from user
transfer_tokens_in(
&ctx.accounts.token_account_a,
&ctx.accounts.token_account_b,
amount_a,
amount_b,
&ctx.accounts.token_program,
&ctx.accounts.owner
)?;
// Update position state
position.pool = ctx.accounts.pool.key();
position.owner = ctx.accounts.owner.key();
position.liquidity = liquidity_delta;
position.tick_lower = tick_lower;
position.tick_upper = tick_upper;
position.bump = ctx.bumps.position;
// Update pool liquidity
update_ticks_liquidity(pool, tick_lower, tick_upper, liquidity_delta, true)?;
Ok(())
}
### 3. `place_limit_order` - Orderbook Integration
Creates a limit order that sits on the order book until matched:
```rust
#[derive(Accounts)]
#[instruction(order_id: u64)]
pub struct PlaceLimitOrder<'info> {
#[account(mut)]
pub owner: Signer<'info>,
#[account(
seeds = [
b"pool",
pool.base_mint.as_ref(),
pool.quote_mint.as_ref()
],
bump = pool.bump
)]
pub pool: Account<'info, Pool>,
#[account(
init,
payer = owner,
space = 8 + LimitOrder::INIT_SPACE,
seeds = [
b"order",
pool.key().as_ref(),
owner.key().as_ref(),
&order_id.to_le_bytes()
],
bump
)]
pub order: Account<'info, LimitOrder>,
#[account(
mut,
token::mint = pool.base_mint,
token::authority = owner
)]
pub user_token_account: Account<'info, TokenAccount>,
#[account(
init,
payer = owner,
token::mint = pool.base_mint,
token::authority = order,
seeds = [
b"order_vault",
order.key().as_ref()
],
bump
)]
pub order_vault: Account<'info, TokenAccount>,
pub token_program: Program<'info, Token>,
pub system_program: Program<'info, System>,
}
#[account]
#[derive(InitSpace)]
pub struct LimitOrder {
pub pool: Pubkey,
pub owner: Pubkey,
pub order_id: u64,
pub tick: i32,
pub amount: u64,
pub is_bid: bool,
pub filled_amount: u64,
pub status: OrderStatus,
pub bump: u8,
}
pub fn place_limit_order(
ctx: Context<PlaceLimitOrder>,
order_id: u64,
tick: i32,
amount: u64,
is_bid: bool,
) -> Result<()> {
let order = &mut ctx.accounts.order;
// Validate tick alignment
let pool = &ctx.accounts.pool;
require!(tick % pool.tick_spacing == 0, ErrorCode::TickNotSpaced);
// Transfer tokens to order vault
let transfer_ctx = CpiContext::new(
ctx.accounts.token_program.to_account_info(),
Transfer {
from: ctx.accounts.user_token_account.to_account_info(),
to: ctx.accounts.order_vault.to_account_info(),
authority: ctx.accounts.owner.to_account_info(),
}
);
transfer(transfer_ctx, amount)?;
// Initialize order
order.pool = ctx.accounts.pool.key();
order.owner = ctx.accounts.owner.key();
order.order_id = order_id;
order.tick = tick;
order.amount = amount;
order.is_bid = is_bid;
order.filled_amount = 0;
order.status = OrderStatus::Open;
order.bump = ctx.bumps.order;
// Emit order placed event
emit!(OrderPlaced {
pool: ctx.accounts.pool.key(),
owner: ctx.accounts.owner.key(),
order_id,
tick,
amount,
is_bid,
timestamp: Clock::get()?.unix_timestamp,
});
Ok(())
}
### 4. `swap` - Execution with Order Matching
Executes a swap, potentially matching against limit orders:
```rust
pub fn swap(
ctx: Context<Swap>,
amount: u64,
sqrt_price_limit: u128,
is_exact_input: bool,
) -> Result<()> {
let pool = &mut ctx.accounts.pool;
// Calculate swap amounts
let (amount_in, amount_out, sqrt_price_new, liquidity) =
compute_swap_step(
pool.sqrt_price,
sqrt_price_limit,
pool.liquidity,
amount,
pool.fee_rate,
is_exact_input
)?;
// Check against limit orders in this tick range
let matched_orders = find_matching_orders(
pool,
pool.current_tick_index,
get_tick_from_sqrt_price(sqrt_price_new),
!is_exact_input // opposite side of trade
);
let mut total_matched = 0;
for order_info in matched_orders {
let order_account = &ctx.remaining_accounts[order_info.index];
let mut order = Account::<LimitOrder>::try_from(order_account)?;
let match_amount = min(order.amount - order.filled_amount, amount_out - total_matched);
// Execute against limit order
execute_against_limit_order(
&mut order,
match_amount,
&ctx.accounts.token_program,
&ctx.accounts.user_token_account,
&ctx.accounts.order_vaults[order_info.vault_index]
)?;
total_matched += match_amount;
if total_matched >= amount_out {
break;
}
}
// Update pool state
pool.sqrt_price = sqrt_price_new;
pool.liquidity = liquidity;
// Apply fees
let protocol_fee = amount_in
.checked_mul(pool.protocol_fee_rate as u64)
.unwrap()
.checked_div(10_000)
.unwrap();
if is_exact_input {
pool.protocol_fees_0 = pool.protocol_fees_0
.checked_add(protocol_fee)
.unwrap();
} else {
pool.protocol_fees_1 = pool.protocol_fees_1
.checked_add(protocol_fee)
.unwrap();
}
Ok(())
}
### 5. `leverage_position` - Leveraged Liquidity
Enables leveraged liquidity provision with up to 5x multiplier:
```rust
pub fn leverage_position(
ctx: Context<LeveragePosition>,
leverage_id: u64,
position_key: Pubkey,
leverage_multiplier: u8,
collateral_amount: u64,
) -> Result<()> {
require!(leverage_multiplier >= 1, ErrorCode::InvalidLeverage);
require!(leverage_multiplier <= MAX_LEVERAGE, ErrorCode::ExceedsMaxLeverage);
let position = &ctx.accounts.position;
let leveraged_position = &mut ctx.accounts.leveraged_position;
// Calculate borrowed amounts
let total_liquidity_value = calculate_position_value(position, ctx.accounts.pool.sqrt_price);
let collateral_value = calculate_token_value(collateral_amount, ctx.accounts.pool.sqrt_price);
let max_borrow_value = collateral_value
.checked_mul(leverage_multiplier as u64)
.unwrap()
.checked_sub(collateral_value)
.unwrap();
// Create leveraged position
leveraged_position.position = position_key;
leveraged_position.owner = ctx.accounts.owner.key();
leveraged_position.leverage_multiplier = leverage_multiplier;
leveraged_position.collateral_amount = collateral_amount;
leveraged_position.borrowed_amount_0 = calculate_borrow_amount_0(total_liquidity_value, max_borrow_value);
leveraged_position.borrowed_amount_1 = calculate_borrow_amount_1(total_liquidity_value, max_borrow_value);
leveraged_position.bump = ctx.bumps.leveraged_position;
// Health check: ensure position remains safe
let health_ratio = calculate_health_ratio(leveraged_position, ctx.accounts.pool.sqrt_price);
require!(health_ratio > MIN_HEALTH_RATIO, ErrorCode::InsufficientCollateral);
Ok(())
}
## Mathematical Formulas
### Concentrated Liquidity Calculations
The amount of token X and Y required for a liquidity position between ticks $t_L$ and $t_U$ is given by:
$$
\Delta x = \Delta L \cdot \left( \frac{1}{\sqrt{P}} - \frac{1}{\sqrt{P_U}} \right)
$$
$$
\Delta y = \Delta L \cdot \left( \sqrt{P} - \sqrt{P_L} \right)
$$
Where:
- $\Delta L$ is the liquidity delta
- $\sqrt{P}$ is the current square root price
- $\sqrt{P_L}, \sqrt{P_U}$ are square root prices at lower and upper ticks
### Swap Computation
For a swap with fee $f$ (in basis points):
Effective amount in after fees:
$$
\Delta x_{eff} = \Delta x \cdot \left(1 - \frac{f}{10^4}\right)
$$
Output amount:
$$
\Delta y = \frac{y \cdot \Delta x_{eff}}{x + \Delta x_{eff}}
$$
### Leverage Health Ratio
$$
\text{Health Ratio} = \frac{\text{Position Value}}{\text{Borrowed Value} \cdot \text{Liquidation Threshold}}
$$
Positions are liquidated when:
$$
\text{Health Ratio} < 1
$$
## Solana & Anchor Best Practices
### 1. Account Validation Patterns
Always validate accounts using Anchor's type system:
```rust
#[account(
constraint = token_account.mint == pool.base_mint,
constraint = token_account.owner == owner.key()
)]
pub token_account: Account<'info, TokenAccount>,
### 2. Compute Unit Optimization
Use iteration limits and batch processing for order matching:
```rust
const MAX_ORDERS_PER_SWAP: usize = 10;
for i in 0..min(remaining_orders.len(), MAX_ORDERS_PER_SWAP) {
// Process order
if compute_units_remaining() < SAFE_COMPUTE_LIMIT {
break;
}
}
### 3. Token-2022 Compatibility
Handle transfer fees by checking received amounts:
```rust
let balance_before = token_account.amount;
transfer(transfer_ctx, amount)?;
let balance_after = token_account.reload()?.amount;
let received_amount = balance_after.checked_sub(balance_before).unwrap();
### 4. Event Emission for Indexers
Emit structured events for easy off-chain processing:
```rust
#[event]
pub struct SwapEvent {
pub pool: Pubkey,
pub trader: Pubkey,
pub amount_in: u64,
pub amount_out: u64,
pub sqrt_price_before: u128,
pub sqrt_price_after: u128,
pub liquidity: u128,
pub timestamp: i64,
}
## Security Considerations
### 1. Access Control
All critical operations use PDA-based authority:
```rust
#[account(
seeds = [b"config"],
bump = config.bump,
constraint = config.admin == admin.key()
)]
pub config: Account<'info, ProtocolConfig>,
### 2. Input Validation
Validate all user inputs with appropriate bounds:
```rust
require!(tick_lower < tick_upper, ErrorCode::InvalidTickRange);
require!(fee_rate <= MAX_FEE_RATE, ErrorCode::InvalidFeeRate);
require!(amount > 0, ErrorCode::ZeroAmount);
### 3. Arithmetic Safety
Use checked arithmetic to prevent overflows:
```rust
let total = amount_a
.checked_add(amount_b)
.ok_or(ErrorCode::ArithmeticOverflow)?;
### 4. Reentrancy Protection
Solana's transaction model prevents reentrancy, but validate cross-program interactions:
```rust
// Ensure token accounts belong to the expected mints
require!(
token_account_a.mint == pool.base_mint &&
token_account_b.mint == pool.quote_mint,
ErrorCode::InvalidTokenAccount
);
### 5. Oracle Manipulation Protection
Use time-weighted prices for sensitive operations:
```rust
let price = calculate_time_weighted_price(
pool.sqrt_price_history,
Clock::get()?.unix_timestamp
);
## How to Use This Contract
### Building and Deploying
```bash
# Build the program
anchor build
# Deploy to devnet
anchor deploy --provider.cluster devnet
# Verify deployment
solana program show --programs
### Example TypeScript Client
```typescript
import * as anchor from "@coral-xyz/anchor";
import { Program } from "@coral-xyz/anchor";
import { DefiTunaAmm } from "../target/types/defi_tuna_amm";
async function createPosition() {
const provider = anchor.AnchorProvider.env();
anchor.setProvider(provider);
const program = anchor.workspace.DefiTunaAmm as Program<DefiTunaAmm>;
const [poolPda] = anchor.web3.PublicKey.findProgramAddressSync(
[
Buffer.from("pool"),
baseMint.toBuffer(),
quoteMint.toBuffer()
],
program.programId
);
const positionId = new anchor.BN(Date.now());
const [positionPda] = anchor.web3.PublicKey.findProgramAddressSync(
[
Buffer.from("position"),
poolPda.toBuffer(),
provider.wallet.publicKey.toBuffer(),
positionId.toArrayLike(Buffer, "le", 8)
],
program.programId
);
const tx = await program.methods
.createPosition(
positionId,
-6000, // tick_lower
6000, // tick_upper
new anchor.BN(1000000) // liquidity
)
.accounts({
pool: poolPda,
position: positionPda,
owner: provider.wallet.publicKey,
tokenAccountA: tokenAccountA,
tokenAccountB: tokenAccountB,
})
.rpc();
console.log("Transaction signature:", tx);
}
### Required Pre-Instructions
For complex operations like leveraged positions, you may need to:
1. Create associated token accounts
2. Approve token transfers
3. Initialize required PDAs
4. Fund accounts with minimum rent
## Extending the Contract
### Adding New Instructions
1. Define new account structs in `#[derive(Accounts)]`
2. Implement handler function with proper validation
3. Add to the `lib.rs` module exports
4. Update IDL generation
### Customization Points
- **Fee Models**: Modify `compute_swap_step` for dynamic fees
- **Order Types**: Extend `LimitOrder` for different order types (FOK, IOC)
- **Leverage Models**: Add new collateral types or liquidation mechanisms
- **Oracle Integration**: Incorporate Pyth or Switchboard for price feeds
### Testing Strategies
```rust
#[tokio::test]
async fn test_swap_with_limit_order_match() {
let mut test = ProgramTest::new(
"defi_tuna_amm",
id(),
processor!(processor::Processor::process)
);
// Add accounts and mints
test.add_account(mint_pubkey, mint_account);
let (mut banks_client, payer, recent_blockhash) = test.start().await;
// Create and place limit order
let place_order_ix = Instruction {
program_id: id(),
accounts: place_order_accounts,
data: place_order_data,
};
// Execute swap that should match
let swap_ix = Instruction {
program_id: id(),
accounts: swap_accounts,
data: swap_data,
};
let transaction = Transaction::new_signed_with_payer(
&[place_order_ix, swap_ix],
Some(&payer.pubkey()),
&[&payer],
recent_blockhash
);
banks_client.process_transaction(transaction).await.unwrap();
}
## Conclusion
The DefiTuna AMM smart contract demonstrates advanced Anchor patterns for building sophisticated DeFi protocols on Solana. By combining concentrated liquidity, limit orders, and leverage in a single program, it showcases how to manage complex state relationships while maintaining security and efficiency.
Key takeaways for developers:
1. Use PDA hierarchies for secure ownership and access control
2. Implement mathematical operations with overflow protection
3. Design for composability with other Solana programs
4. Emit comprehensive events for off-chain indexing
5. Optimize for compute units in iteration-heavy operations
This contract serves as a foundation for building next-generation AMMs that bridge the gap between traditional order books and automated market makers.
TL;DR: Production swap execution engine with guardrails, operational controls, and analytics-ready telemetry.
Jupiter aggregates liquidity from 20+ venues (CPAMM, CLMM, DLMM, PMM, order books) providing best execution, reduced slippage, and MEV protection through automatic routing and order splitting.
Direct Jupiter API usage is simple, but wrapping it in an Anchor program enables:
| Feature | Direct API | Anchor program wrapper |
|---|---|---|
| Fee collection | manual logic | on-chain enforcement |
| Platform branding | client-side only | program-owned config |
| Access control | none | admin-gated pause/update |
| Composability | limited | CPI-friendly for other protocols |
| Audit trail | off-chain | on-chain events |
| Slippage protection | client-side | program-enforced |
Understanding enforcement boundaries is critical for security and UX:
| Concern | Enforced on-chain | Enforced in client | Notes |
|---|---|---|---|
| Slippage ceiling | Yes | Yes | On-chain cap prevents hostile clients from bypassing limits |
| Fee collection | Yes | No | Must be deterministic; client cannot skip or reduce fees |
| Quote freshness | No | Yes | Client refreshes quotes; include quote timestamp in intent |
| Route allowlist/denylist | Optional | Optional | Useful for risk control (e.g., block suspicious pools) |
| Pause / emergency stop | Yes | No | Admin can halt swaps immediately for incidents |
| Compute budget | No | Yes | Client requests higher compute units for complex routes |
| Intent deduplication | No | Off-chain | Backend checks intent_id before indexing |
Key principle: On-chain enforces non-bypassable invariants (fees, caps, pauses). Client enforces UX optimizations (quote refresh, compute). Off-chain systems handle analytics and deduplication.
This swap integration is a component in an operating system, not a standalone feature. Understanding the full lifecycle is critical for product reliability:
Workflow: UI → Quote → Intent (idempotency key) → Execute (policy enforcement) → Telemetry (event emission) → Indexer (DB write) → Backoffice (dashboards, support)
Benefits: Idempotency prevents double-swaps, attribution tracks user sessions, debuggability via intent IDs, operability through conversion dashboards.
In production systems, you need reliable execution and clean analytics. The intent model provides both.
A SwapIntent captures user action before execution:
interface SwapIntent {
intent_id: string; // UUIDv4 or client-generated unique ID
wallet: PublicKey; // User wallet address
input_mint: PublicKey; // Source token
output_mint: PublicKey; // Destination token
amount_in: u64; // Input amount (lamports)
max_slippage_bps: u16; // Max acceptable slippage
created_at: i64; // Client timestamp
client_version: string; // App version (e.g., "web-v2.1.3")
metadata?: Record<string, string>; // Campaign ID, referrer, etc.
}
import { v4 as uuidv4 } from 'uuid';
function createSwapIntent(params: {
wallet: PublicKey;
inputMint: PublicKey;
outputMint: PublicKey;
amountIn: number;
slippageBps: number;
}): SwapIntent {
return {
intent_id: uuidv4(),
wallet: params.wallet,
input_mint: params.inputMint,
output_mint: params.outputMint,
amount_in: params.amountIn,
max_slippage_bps: params.slippageBps,
created_at: Date.now(),
client_version: process.env.NEXT_PUBLIC_APP_VERSION || 'unknown',
metadata: {
campaign_id: getCampaignId(),
referrer: document.referrer,
},
};
}
1. Idempotency (no double-swaps):
// User clicks "Swap" → creates intent
const intent = createSwapIntent(params);
// Network error on first attempt
try {
await executeSwap(intent); // Fails
} catch (error) {
// User retries with SAME intent_id
await executeSwap(intent); // Succeeds
}
// Backend deduplicates by intent_id
// Only ONE swap executed, even with multiple transactions
2. Consistent analytics: Every event has intent_id enabling quote-to-swap conversion tracking and funnel analysis.
3. Clean support: Intent IDs allow instant lookup of swap context (stale quote, slippage, route issues).
Storage: Store intents client-side (localStorage), off-chain DB (on initiation), and in events (correlation).
programs/token-swap/src/
├── state.rs # Account structures
│ └── JupiterConfig # 78 bytes: admin, fees, slippage, pause
├── constants.rs # Program IDs and seeds
│ ├── JUPITER_V6_PROGRAM_ID # JUP6LkbZbjS1jKKwapdHNy74zcZ3tLUZoi5QNyVTaV4
│ └── JUPITER_CONFIG_SEED # "jupiter_config"
├── errors.rs # 8 custom error types
│ ├── InvalidAmount
│ ├── JupiterPaused
│ └── MinimumOutputNotMet
└── instructions/
├── init_jupiter_config.rs # Initialize config PDA
├── update_jupiter_config.rs # Admin updates
├── jupiter_swap.rs # Main swap execution
└── jupiter_route_swap.rs # Legacy route support
#[account]
pub struct JupiterConfig {
pub admin: Pubkey, // 32 bytes
pub fee_account: Pubkey, // 32 bytes
pub platform_fee_bps: u16, // 2 bytes (0-10000)
pub max_slippage_bps: u16, // 2 bytes (0-10000)
pub paused: bool, // 1 byte
pub bump: u8, // 1 byte
} // Total: 70 bytes (+ 8 discriminator = 78)
Design decisions:
u16 for BPS values (supports full 0-10000 range, 100% = 10000 BPS)#[derive(Accounts)]
pub struct InitJupiterConfig<'info> {
#[account(
init,
payer = admin,
space = 8 + 70,
seeds = [b"jupiter_config"],
bump
)]
pub config: Account<'info, JupiterConfig>,
#[account(mut)]
pub admin: Signer<'info>,
pub system_program: Program<'info, System>,
}
pub fn init_jupiter_config(
ctx: Context<InitJupiterConfig>,
fee_account: Pubkey,
platform_fee_bps: u16,
max_slippage_bps: u16,
) -> Result<()> {
require!(
platform_fee_bps <= 1000,
JupiterSwapError::InvalidPlatformFee
);
require!(
max_slippage_bps <= 10000,
JupiterSwapError::InvalidMaxSlippage
);
let config = &mut ctx.accounts.config;
config.admin = ctx.accounts.admin.key();
config.fee_account = fee_account;
config.platform_fee_bps = platform_fee_bps;
config.max_slippage_bps = max_slippage_bps;
config.paused = false;
config.bump = ctx.bumps.config;
Ok(())
}
Key validations:
#[derive(Accounts)]
pub struct JupiterSwap<'info> {
#[account(
seeds = [b"jupiter_config"],
bump = config.bump
)]
pub config: Account<'info, JupiterConfig>,
#[account(mut)]
pub user: Signer<'info>,
/// CHECK: Jupiter v6 program ID verified in instruction
pub jupiter_program: UncheckedAccount<'info>,
// Token accounts + remaining accounts for Jupiter routing
}
pub fn jupiter_swap<'info>(
ctx: Context<'_, '_, 'info, 'info, JupiterSwap<'info>>,
amount_in: u64,
minimum_amount_out: u64,
) -> Result<()> {
let config = &ctx.accounts.config;
// 1. Validate state
require!(!config.paused, JupiterSwapError::JupiterPaused);
require!(amount_in > 0, JupiterSwapError::InvalidAmount);
// 2. Verify Jupiter program ID
require_keys_eq!(
ctx.accounts.jupiter_program.key(),
JUPITER_V6_PROGRAM_ID.parse::<Pubkey>().unwrap(),
JupiterSwapError::InvalidJupiterProgram
);
// 3. Build CPI accounts (dynamically from remaining_accounts)
let mut accounts = Vec::new();
for account in ctx.remaining_accounts.iter() {
accounts.push(AccountMeta {
pubkey: *account.key,
is_signer: account.is_signer,
is_writable: account.is_writable,
});
}
// 4. Execute CPI to Jupiter
let swap_ix = Instruction {
program_id: ctx.accounts.jupiter_program.key(),
accounts,
data: swap_data, // Jupiter swap instruction data
};
invoke_signed(&swap_ix, ctx.remaining_accounts, &[])?;
// 5. Verify output amount (Token-2022 safe)
let output_amount = observe_vault_delta(); // observe vault delta
require!(
output_amount >= minimum_amount_out,
JupiterSwapError::MinimumOutputNotMet
);
// 6. Collect platform fee
if config.platform_fee_bps > 0 {
let fee = (output_amount as u128)
.checked_mul(config.platform_fee_bps as u128)
.unwrap()
.checked_div(10000)
.unwrap() as u64;
// Transfer fee to platform account
}
// 7. Emit event for analytics
emit!(JupiterSwapEvent {
user: ctx.accounts.user.key(),
amount_in,
amount_out: output_amount,
platform_fee: fee,
});
Ok(())
}
Critical implementation details:
// WRONG: Trusting instruction amount
let output_amount = minimum_amount_out;
// CORRECT: Observe vault delta
let vault_before = user_output_token.amount;
// ... execute swap ...
user_output_token.reload()?;
let output_amount = user_output_token.amount
.saturating_sub(vault_before);
Transfer fees and hooks mean you cannot trust amounts in instruction data.
Jupiter requires dynamic account lists (routes vary by liquidity):
// Frontend passes all necessary accounts
const remainingAccounts = [
{ pubkey: userSourceToken, isSigner: false, isWritable: true },
{ pubkey: userDestToken, isSigner: false, isWritable: true },
// ... all intermediary pool accounts from Jupiter API
];
Program must accept remaining_accounts via:
pub fn jupiter_swap<'info>(
ctx: Context<'_, '_, 'info, 'info, JupiterSwap<'info>>,
// ^^^ lifetime annotation required for remaining_accounts
pub fn update_jupiter_config(
ctx: Context<UpdateJupiterConfig>,
new_admin: Option<Pubkey>,
new_fee_account: Option<Pubkey>,
new_platform_fee_bps: Option<u16>,
new_max_slippage_bps: Option<u16>,
new_paused: Option<bool>,
) -> Result<()> {
let config = &mut ctx.accounts.config;
// Validate admin
require_keys_eq!(
ctx.accounts.admin.key(),
config.admin,
JupiterSwapError::Unauthorized
);
// Optional updates (partial update pattern)
if let Some(admin) = new_admin {
config.admin = admin;
}
if let Some(fee_bps) = new_platform_fee_bps {
require!(fee_bps <= 1000, JupiterSwapError::InvalidPlatformFee);
config.platform_fee_bps = fee_bps;
}
// ... other optional fields
Ok(())
}
Partial update pattern: All fields optional → supports single-field updates without re-specifying everything.
import { useConnection, useWallet } from '@solana/wallet-adapter-react';
import { PublicKey, VersionedTransaction } from '@solana/web3.js';
import { useProgram } from './useSwapProgram';
export function useJupiter() {
const { connection } = useConnection();
const wallet = useWallet();
const program = useProgram();
async function getQuote(params: {
inputMint: string;
outputMint: string;
amount: number;
slippageBps?: number;
}) {
const response = await fetch(
`https://quote-api.jup.ag/v6/quote?` +
new URLSearchParams({
inputMint: params.inputMint,
outputMint: params.outputMint,
amount: params.amount.toString(),
slippageBps: (params.slippageBps || 50).toString(),
})
);
return response.json();
}
async function getSwapInstructions(quote: any) {
const response = await fetch('https://quote-api.jup.ag/v6/swap-instructions', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
quoteResponse: quote,
userPublicKey: wallet.publicKey!.toBase58(),
wrapAndUnwrapSol: true,
// Use versioned transactions for ALT support
useVersionedTransaction: true,
}),
});
return response.json();
}
async function executeSwapWithProgram(quote: any) {
if (!wallet.publicKey || !program) return;
const { swapInstruction } = await getSwapInstructions(quote);
// Get config PDA
const [configPda] = PublicKey.findProgramAddressSync(
[Buffer.from('jupiter_config')],
program.programId
);
// Build transaction via Anchor
const tx = await program.methods
.jupiterSwap(
new BN(quote.inAmount),
new BN(quote.outAmount)
)
.accounts({
config: configPda,
user: wallet.publicKey,
jupiterProgram: new PublicKey('JUP6LkbZbjS1jKKwapdHNy74zcZ3tLUZoi5QNyVTaV4'),
})
.remainingAccounts(swapInstruction.accounts) // Dynamic routing accounts
.transaction();
// Handle Address Lookup Tables if present
if (swapInstruction.addressLookupTableAccounts?.length > 0) {
const lookupTables = await Promise.all(
swapInstruction.addressLookupTableAccounts.map((key: string) =>
connection.getAddressLookupTable(new PublicKey(key))
)
);
// Build versioned transaction
const message = new TransactionMessage({
payerKey: wallet.publicKey,
recentBlockhash: (await connection.getLatestBlockhash()).blockhash,
instructions: tx.instructions,
}).compileToV0Message(lookupTables.map(lt => lt.value!));
const versionedTx = new VersionedTransaction(message);
await wallet.sendTransaction(versionedTx, connection);
} else {
await wallet.sendTransaction(tx, connection);
}
}
return { getQuote, executeSwapWithProgram };
}
Key frontend considerations:
Complex Jupiter routes exceed the 1232-byte transaction limit. ALTs compress account lists:
// Without ALT: 32 bytes per account × 40 accounts = 1280 bytes (fails)
// With ALT: table reference + indices = ~50 bytes (works)
Use versioned transactions (v0) to support ALTs.
Jupiter quotes expire quickly (10-30 seconds):
const quote = await getQuote(params);
// Wait too long...
await sleep(60000); // Quote now stale
await executeSwap(quote); // Likely fails with slippage error
Best practice: poll quotes every 5-10 seconds during user review.
'use client';
import { useState, useEffect } from 'react';
import { useJupiter } from '@/hooks/useJupiter';
const TOKENS = {
SOL: 'So11111111111111111111111111111111111111112',
USDC: 'EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v',
USDT: 'Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB',
JUP: 'JUPyiwrYJFskUPiHa7hkeR8VUtAeFoSYbKedZNsDvCN',
};
export function JupiterSwap() {
const { getQuote, executeSwapWithProgram } = useJupiter();
const [inputMint, setInputMint] = useState(TOKENS.SOL);
const [outputMint, setOutputMint] = useState(TOKENS.USDC);
const [amount, setAmount] = useState('1.0');
const [quote, setQuote] = useState<any>(null);
const [loading, setLoading] = useState(false);
// Auto-refresh quote every 10 seconds
useEffect(() => {
const interval = setInterval(async () => {
if (amount && parseFloat(amount) > 0) {
const q = await getQuote({
inputMint,
outputMint,
amount: parseFloat(amount) * 1e9, // Convert to lamports
slippageBps: 50,
});
setQuote(q);
}
}, 10000);
return () => clearInterval(interval);
}, [inputMint, outputMint, amount]);
const handleSwap = async () => {
setLoading(true);
try {
await executeSwapWithProgram(quote);
// Success notification
} catch (error) {
console.error('Swap failed:', error);
} finally {
setLoading(false);
}
};
return (
<div className="swap-card">
<div className="input-section">
<input
type="number"
value={amount}
onChange={(e) => setAmount(e.target.value)}
placeholder="Amount"
/>
<select value={inputMint} onChange={(e) => setInputMint(e.target.value)}>
<option value={TOKENS.SOL}>SOL</option>
<option value={TOKENS.USDC}>USDC</option>
<option value={TOKENS.USDT}>USDT</option>
<option value={TOKENS.JUP}>JUP</option>
</select>
</div>
<div className="output-section">
<div className="estimated-output">
{quote ? (
<>
<span className="amount">{(quote.outAmount / 1e6).toFixed(6)}</span>
<span className="price-impact">
Price impact: {(quote.priceImpactPct * 100).toFixed(2)}%
</span>
</>
) : (
<span className="loading">Fetching quote...</span>
)}
</div>
<select value={outputMint} onChange={(e) => setOutputMint(e.target.value)}>
<option value={TOKENS.USDC}>USDC</option>
<option value={TOKENS.USDT}>USDT</option>
<option value={TOKENS.SOL}>SOL</option>
<option value={TOKENS.JUP}>JUP</option>
</select>
</div>
<button
onClick={handleSwap}
disabled={loading || !quote}
className="swap-button"
>
{loading ? 'Swapping...' : 'Swap'}
</button>
{quote && (
<div className="route-info">
<div>Route: {quote.routePlan?.map((r: any) => r.swapInfo.label).join(' → ')}</div>
<div>Min output: {((quote.outAmount * 0.995) / 1e6).toFixed(6)} (0.5% slippage)</div>
</div>
)}
</div>
);
}
UX enhancements:
| Check | Why it matters | How to verify |
|---|---|---|
| Program ID fixed | Reproducible builds | anchor build --verifiable |
| Upgrade authority | Immutability post-audit | solana program set-upgrade-authority <program_id> --final |
| Config admin | Emergency controls | Multisig or DAO-controlled |
| Platform fee ≤ 1% | Competitive with alternatives | Review platform_fee_bps |
| Max slippage reasonable | Protect users | Typically 50-500 BPS |
| Pause mechanism tested | Kill-switch works | Integration test coverage |
| Token-2022 tested | Fee-on-transfer handling | Test with USDT (transfer fees) |
# 1. Build verifiable program
anchor build --verifiable
# 2. Deploy to devnet
anchor deploy --provider.cluster devnet
# 3. Initialize config (via multisig in production)
anchor run initialize-config --provider.cluster devnet
# 4. Verify deployment
solana program show <program_id>
# 5. Audit & security review
# (Use Sec3, OtterSec, or similar)
# 6. Deploy to mainnet
anchor deploy --provider.cluster mainnet-beta
# 7. Initialize mainnet config
anchor run initialize-config --provider.cluster mainnet-beta
# 8. Set upgrade authority to final
solana program set-upgrade-authority <program_id> --final
When swaps fail or users contact support, you need immediate answers. This runbook maps symptoms to root causes.
Every support ticket needs:
| Symptom | Root cause | Diagnosis | Immediate mitigation |
|---|---|---|---|
| "Slippage tolerance exceeded" | Quote stale OR low liquidity | Check quote_age_seconds in event | Shorten quote refresh interval to 5-10s |
| "Transaction simulation failed" | Compute budget exceeded | Check route complexity (hops >3) | Bump compute units to 400k for complex routes |
| "Account not found" | ALT missing or not loaded | Check addressLookupTableAccounts in txn | Ensure ALTs created and extended with pool accounts |
| "Insufficient funds" | User balance < amount + fees | Check wallet balance vs amount_in | Show clear error: "Need X SOL for fees" |
| "Custom program error: 0x1770" | Token-2022 transfer fee | Check if token has transfer fee extension | Use vault delta verification (not instruction data) |
| "Transaction timeout" | Network congestion | Check priority fee paid | Increase priority fee dynamically (use Helius API) |
| "Invalid instruction data" | Jupiter program upgraded | Check program version mismatch | Update Jupiter program ID constant |
| Swap succeeds but user didn't receive full amount | Token-2022 fee-on-transfer | Compare quote vs actual received | Document this in UI ("Receives ~X after fees") |
Immediate actions:
Check Solana network status (https://status.solana.com)
Verify Jupiter API health (https://status.jup.ag)
Query last 100 failures by error_code:
SELECT error_code, COUNT(*) as count
FROM swap_failed_events
WHERE timestamp > NOW() - INTERVAL '1 hour'
GROUP BY error_code
ORDER BY count DESC;
If error_code=1005 (compute exceeded): Bump compute budget globally
If error_code=1002 (stale quote): Reduce quote refresh interval
If widespread network issue: Enable pause toggle via admin
Debug flow:
JupiterSwapEvent.amount_outJupiterSwapFailedEvent.error_codeDiagnostic queries:
-- Check if swap volume dropped or fee collection failed
SELECT
DATE_TRUNC('hour', timestamp) as hour,
COUNT(*) as swap_count,
SUM(amount_in) as total_volume,
SUM(platform_fee) as revenue
FROM swap_events
WHERE timestamp > NOW() - INTERVAL '24 hours'
GROUP BY hour
ORDER BY hour DESC;
Possible causes:
Set up alerts for:
Problem: Complex Jupiter routes require 30-50 accounts, exceeding transaction limits.
Solution: Use Address Lookup Tables (ALTs)
// Create ALT during program initialization
const [lookupTable, _] = AddressLookupTableProgram.createLookupTable({
authority: admin,
payer: admin,
recentSlot: await connection.getSlot(),
});
// Add frequently-used accounts
await connection.sendTransaction(
AddressLookupTableProgram.extendLookupTable({
lookupTable,
authority: admin,
payer: admin,
addresses: [USDC_MINT, USDT_MINT, /* common pools */],
})
);
Problem: Mainnet has higher volatility and MEV, causing more slippage failures.
Solution: Dynamic slippage based on liquidity
function calculateSlippage(quote: JupiterQuote) {
const priceImpact = quote.priceImpactPct;
if (priceImpact < 0.01) return 50; // 0.5% for deep liquidity
if (priceImpact < 0.05) return 100; // 1% for medium liquidity
return 500; // 5% for low liquidity
}
Problem: USDT (SPL Token-2022 with transfer fees) results in less than expected amounts.
Solution: Always observe vault deltas
let before = token_account.amount;
// ... execute transfer ...
token_account.reload()?;
let actual_received = token_account.amount.saturating_sub(before);
Problem: User reviews swap for 60 seconds, quote becomes stale, transaction fails.
Solution: Auto-refresh quotes
useEffect(() => {
const interval = setInterval(refreshQuote, 10000); // Every 10s
return () => clearInterval(interval);
}, [inputMint, outputMint, amount]);
Problem: Complex routes run out of compute units (200k default).
Solution: Request higher compute budget
use solana_program::compute_budget::ComputeBudgetInstruction;
// Add as first instruction in transaction
let compute_ix = ComputeBudgetInstruction::set_compute_unit_limit(400_000);
// Inefficient: Multiple account lookups
for account in ctx.remaining_accounts.iter() {
let data = account.try_borrow_data()?;
// Process...
}
// Efficient: Single borrow per account
let accounts: Vec<_> = ctx.remaining_accounts
.iter()
.map(|a| (a.key(), a.try_borrow_data()))
.collect();
// Slower: Division
let fee = (amount * fee_bps) / 10000;
// Faster: Shift (if fee_bps is power of 2)
// For 0.5% (50 bps): multiply by 1/200 = right shift by ~8
// Not always applicable, but pattern to consider
// Sequential quotes (slower)
const quote1 = await getQuote({ inputMint: SOL, outputMint: USDC, amount: 1e9 });
const quote2 = await getQuote({ inputMint: SOL, outputMint: USDT, amount: 1e9 });
// Parallel quotes (faster)
const [quote1, quote2] = await Promise.all([
getQuote({ inputMint: SOL, outputMint: USDC, amount: 1e9 }),
getQuote({ inputMint: SOL, outputMint: USDT, amount: 1e9 }),
]);
Your event schema defines what you can measure and debug. Make it comprehensive:
#[event]
pub struct JupiterSwapEvent {
// Attribution
pub intent_id: [u8; 16], // UUID bytes for client-side correlation
pub user: Pubkey, // Wallet address
pub client_version: [u8; 32], // App version (e.g., "web-v2.1.3\0\0...")
// Swap details
pub input_mint: Pubkey,
pub output_mint: Pubkey,
pub amount_in: u64,
pub amount_out: u64,
pub platform_fee: u64,
// Execution context
pub quote_timestamp: i64, // When quote was generated (detect staleness)
pub execution_timestamp: i64, // When swap executed
pub route_hash: u64, // Hash of route plan (fingerprint venues used)
pub slippage_bps_requested: u16, // User-requested slippage
pub slippage_bps_effective: u16, // Actual slippage observed
// Operational data
pub compute_units_consumed: u64, // For performance tuning
pub priority_fee_paid: u64, // MEV/congestion analysis
}
#[event]
pub struct JupiterSwapFailedEvent {
// Attribution (same as success event)
pub intent_id: [u8; 16],
pub user: Pubkey,
pub client_version: [u8; 32],
// Failure context
pub input_mint: Pubkey,
pub output_mint: Pubkey,
pub amount_in: u64,
pub minimum_amount_out: u64,
// Diagnostic data
pub error_code: u32, // Mapped to human-readable reasons
pub program_error: Option<String>, // Anchor error details
pub quote_age_seconds: i64, // How old was the quote?
pub timestamp: i64,
}
Why these fields matter:
intent_id: Join client logs, backend DB, and on-chain events for full traceclient_version: Identify bugs introduced in specific releasesquote_timestamp vs execution_timestamp: Measure latency, detect stale quotesroute_hash: Identify which venue combinations succeed/fail mostslippage_bps_effective: Measure if users are over-allocating slippage tolerancecompute_units_consumed: Optimize compute budgets dynamicallyerror_code: Build dashboards showing top failure reasonsMost teams only emit success events. Failures are more valuable for ops:
pub fn jupiter_swap<'info>(
ctx: Context<'_, '_, 'info, 'info, JupiterSwap<'info>>,
intent_id: [u8; 16],
amount_in: u64,
minimum_amount_out: u64,
) -> Result<()> {
// ... validation and swap execution ...
// If swap fails, emit failure event BEFORE returning error
if let Err(e) = execute_jupiter_cpi(&ctx, amount_in) {
emit!(JupiterSwapFailedEvent {
intent_id,
user: ctx.accounts.user.key(),
client_version: get_client_version(),
input_mint: ctx.accounts.input_mint.key(),
output_mint: ctx.accounts.output_mint.key(),
amount_in,
minimum_amount_out,
error_code: map_error_to_code(&e),
program_error: Some(e.to_string()),
quote_age_seconds: calculate_quote_age(),
timestamp: Clock::get()?.unix_timestamp,
});
return Err(e);
}
// Success: emit success event
emit!(JupiterSwapEvent { /* ... */ });
Ok(())
}
Error code mapping (for structured dashboards):
fn map_error_to_code(error: &anchor_lang::error::Error) -> u32 {
match error {
JupiterSwapError::MinimumOutputNotMet => 1001,
JupiterSwapError::StaleQuote => 1002,
JupiterSwapError::JupiterPaused => 1003,
JupiterSwapError::InsufficientLiquidity => 1004,
JupiterSwapError::ComputeBudgetExceeded => 1005,
JupiterSwapError::ALTMissing => 1006,
JupiterSwapError::Token2022TransferFee => 1007,
_ => 9999, // Unknown error
}
}
These metrics map directly to SQL queries and BI dashboards:
| KPI | Formula | Query example | Target |
|---|---|---|---|
| Quote → Swap conversion | completed_swaps / quote_requests | SELECT COUNT(DISTINCT intent_id) FROM swaps / COUNT(*) FROM quotes | greater than 80% |
| Intent completion rate | completed_intents / created_intents | SELECT successful / total FROM intent_summary | greater than 90% |
| Failure rate by reason | failures(reason=X) / total_attempts | SELECT error_code, COUNT(*) / total FROM failures GROUP BY error_code | less than 5% overall |
| Median execution latency | median(event_time - intent_creation_time) | SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY latency) FROM swaps | less than 15s |
| Revenue (USD) | SUM(platform_fee * token_price_usd) | SELECT SUM(f.amount * p.price) FROM fees f JOIN prices p ON f.mint = p.mint | Track growth |
| Route health score | success_rate_per_route_fingerprint | SELECT route_hash, COUNT(*) successes / total FROM swaps GROUP BY route_hash | greater than 95% per route |
| Effective slippage | AVG((quote_out - actual_out) / quote_out * 10000) | SELECT AVG((quoted - actual) / quoted * 10000) FROM swaps | less than 50 BPS |
| Repeat user rate | users_with_2plus_swaps / total_users | SELECT COUNT(DISTINCT user) FROM (SELECT user, COUNT(*) c FROM swaps GROUP BY user HAVING c >= 2) | greater than 40% |
Sample dashboard SQL (Postgres):
-- Real-time conversion funnel
WITH funnel AS (
SELECT
COUNT(DISTINCT q.intent_id) as quotes,
COUNT(DISTINCT s.intent_id) as swaps,
COUNT(DISTINCT CASE WHEN s.success = true THEN s.intent_id END) as successful
FROM quote_requests q
LEFT JOIN swap_events s ON q.intent_id = s.intent_id
WHERE q.created_at > NOW() - INTERVAL '1 hour'
)
SELECT
quotes,
swaps,
successful,
ROUND(100.0 * swaps / NULLIF(quotes, 0), 2) as quote_to_swap_pct,
ROUND(100.0 * successful / NULLIF(swaps, 0), 2) as success_rate
FROM funnel;
-- Top failure reasons (last 24 hours)
SELECT
CASE error_code
WHEN 1001 THEN 'Slippage exceeded'
WHEN 1002 THEN 'Stale quote'
WHEN 1005 THEN 'Compute budget exceeded'
WHEN 1006 THEN 'ALT missing'
WHEN 1007 THEN 'Token-2022 transfer fee'
ELSE 'Unknown'
END as failure_reason,
COUNT(*) as occurrences,
ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 2) as percentage
FROM swap_failed_events
WHERE timestamp > NOW() - INTERVAL '24 hours'
GROUP BY error_code
ORDER BY occurrences DESC;
-- Revenue by token pair (last 7 days)
SELECT
s.input_mint,
s.output_mint,
COUNT(*) as swap_count,
SUM(s.platform_fee) as total_fee_tokens,
SUM(s.platform_fee * p.price_usd) as revenue_usd
FROM swap_events s
JOIN token_prices p ON s.output_mint = p.mint
WHERE s.timestamp > NOW() - INTERVAL '7 days'
GROUP BY s.input_mint, s.output_mint
ORDER BY revenue_usd DESC
LIMIT 10;
Grafana/Metabase integration:
// Subscribe to program logs
const connection = new Connection(HELIUS_RPC_URL);
connection.onLogs(
programId,
(logs) => {
if (logs.logs.some(log => log.includes('JupiterSwapEvent'))) {
// Parse event and store in database
const event = parseJupiterSwapEvent(logs);
await db.swaps.insert(event);
}
},
'confirmed'
);
| Operation | Compute units | Typical cost (SOL) | Notes |
|---|---|---|---|
| Init config | ~5,000 | 0.000005 | One-time |
| Simple swap (1 hop) | ~50,000 | 0.00005 | Direct pool |
| Complex swap (3+ hops) | ~200,000 | 0.0002 | Multi-route |
| Update config | ~3,000 | 0.000003 | Admin only |
Priority fees: Add 0.00001-0.0001 SOL during congestion for faster inclusion.
Example with 0.3% platform fee:
At 1M SOL monthly volume (current mid-tier DEX):
Jupiter DCA allows scheduled swaps:
// Create DCA order
const dcaOrder = await program.methods
.createDcaOrder({
inputMint: SOL,
outputMint: USDC,
amountPerSwap: 10 * 1e9, // 10 SOL
interval: 3600, // 1 hour
totalAmount: 1000 * 1e9, // 1000 SOL total
})
.rpc();
Integrate Jupiter's private RPC:
const quote = await getQuote({
inputMint,
outputMint,
amount,
// Use private RPC to avoid frontrunning
rpcUrl: 'https://private.jup.ag/rpc',
});
Jupiter integrates Wormhole for cross-chain swaps:
pub fn cross_chain_swap(
ctx: Context<CrossChainSwap>,
destination_chain: u16, // e.g., 1 = Ethereum
amount: u64,
) -> Result<()> {
// Swap SOL → USDC on Solana
// Bridge USDC to Ethereum via Wormhole
// Swap USDC → ETH on Ethereum
Ok(())
}
Show users why Jupiter found better price:
interface RouteBreakdown {
venue: string;
percentage: number; // % of trade routed through this venue
priceImpact: number;
}
const breakdown: RouteBreakdown[] = [
{ venue: 'Raydium CLMM', percentage: 60, priceImpact: 0.12 },
{ venue: 'Orca Whirlpool', percentage: 30, priceImpact: 0.08 },
{ venue: 'Meteora DLMM', percentage: 10, priceImpact: 0.05 },
];
| Aspect | Direct API | Program wrapper (this guide) |
|---|---|---|
| Implementation time | 2 hours | 1 week (with tests) |
| Fee collection | manual off-chain | automatic on-chain |
| Composability | limited | full CPI support |
| Audit surface | none | program code |
| Branding | client-side only | on-chain enforcement |
| Access control | none | admin-gated |
| Event logs | none | on-chain events |
| Upgrade path | N/A | versioned program |
When to use direct API:
When to use program wrapper:
Use this checklist before deploying to mainnet:
quote_timestamp on backend before executionchecked_mul / checked_divSwapFailedEvent with error_code before returning errorsanchor build --verifiable succeeds; hash matches deployed program--final after auditBuilding a Jupiter integration on Solana requires:
The result is a composable, fee-collecting swap infrastructure that leverages Jupiter's best-in-class routing while maintaining control over user experience and revenue.
Key takeaways:
Ship safe.
An AMM is any on-chain market maker that:
An on-chain order book can be fully on-chain too, but it’s not an AMM: it matches explicit bids/asks, not a curve/invariant rule.
A CPAMM is a constant-function AMM where:
x and y are pool reserves.
So:
Let reserves be (x, y) and you swap dx of X for Y.
If fee is f (e.g. 0.003 for 30 bps):
x := x + dxy := y - dyObserved vault delta (Token-2022-safe input amount):
p ≈ y/x (direction depends on quote convention)k over time (LPs get paid via reserve growth).| Term / design | Category? | Core idea | Typical on-chain state | Who provides liquidity? | Quote source |
|---|---|---|---|---|---|
| AMM | Yes | Algorithmic pricing vs state | varies | varies | curve/parameters/inventory |
| CFAMM (constant-function AMM) | Yes | Trades move along an invariant | reserves + params | LPs or protocol | invariant |
| CPAMM / CPMM | Yes | x*y=k | 2 vaults + pool state (+ LP mint) | passive LPs | reserves ratio |
| StableSwap | Yes | hybrid curve (sum-like near peg) | vaults + params (A, etc.) | passive LPs | curve + params |
| CLMM | Yes | liquidity concentrated in ranges/ticks | vaults + tick arrays + position accounts | active LPs | ticks + reserves |
| DLMM (bins) | Yes | discrete bins + liquidity distribution | vaults + bin arrays + position state | active/semi-active LPs | bins + params |
| PMM / oracle-anchored | Yes | price anchored to oracle fair value | inventory + params + oracle feeds | market maker / protocol | oracle + model |
| CLOB (order book) | No (not AMM) | match bids/asks | market + order state | makers | limit orders |
| TWAMM | No (mechanism) | execute large order over time | long-term order state | trader orders | schedule |
| Bonding curve launch | Yes (often) | virtual reserves / issuance curve | curve params + reserves | launch pool | curve |
| Design | Typical spread / slippage (for same TVL) | “Always liquid”? | Best for | Pain points (trader) | Router friendliness |
|---|---|---|---|---|---|
| CPAMM | worst for tight markets | Yes | long-tail discovery, simple swaps | high price impact without huge TVL | high (simple routes) |
| StableSwap | excellent near peg | Yes (until extreme imbalance) | stable/stable, correlated assets | parameter risk; off-peg behavior | high |
| CLMM | best near spot | No (can be out-of-range) | majors, low slippage | depth depends on LP ranges | high (but more accounts) |
| DLMM | very good when bins are well-set | mostly | structured liquidity & dynamic fees | bin distribution matters | high (but more accounts) |
| PMM | potentially excellent | depends on MM inventory | majors & flow-driven quoting | oracle/model risk; opaque behavior | high if integrated (RFQ-like) |
| CLOB | best when book is thick | n/a | pro trading, limit orders | needs makers & incentives | medium/high (depends on infra) |
| TWAMM | optimized for large orders | n/a | size execution | not instant | routed as a strategy leg |
| Bonding curve | deterministic but can be harsh | curve-dependent | launches | can be gamed / MEV-heavy | usually “launch-only” |
| Design | LP position type | Capital efficiency | IL profile | Operational complexity | Who tends to outperform? |
|---|---|---|---|---|---|
| CPAMM | fungible LP token | low | classic IL (full range) | low | passive LPs in long-tail / high fees |
| StableSwap | often fungible LP | high near peg | smaller IL near peg | medium | LPs in correlated pairs |
| CLMM | tokenized/NFT-like position | very high | can be worse if misranged | high | sophisticated LPs / managed vaults |
| DLMM | bin/strategy position state | high (configurable) | strategy-dependent | medium/high | strategy LPs; can be “MM-like” |
| PMM | usually MM-managed inventory | high | model-controlled | high | market makers (not passive LPs) |
| CLOB | maker orders | n/a | inventory risk, not IL | high | professional makers |
| Bonding curve | not traditional LP | n/a | n/a | medium | launch designers + snipers (unless mitigated) |
This is the table people skip, but it often determines what scales.
| Design | What gets written per swap? | Hot-account tendency | Parallelism shape | Tx/account footprint | Notes |
|---|---|---|---|---|---|
| CPAMM | same pool state + both vaults | high | many swaps serialize on same pool | low/medium | simplest, but hotspot-prone |
| StableSwap | same as CP-ish + params | high | similar to CP contention | medium | more compute than CP |
| CLMM | vaults + tick arrays + position-related state | medium | can shard via tick arrays | higher | more accounts; better scaling shape |
| DLMM | vaults + active bin(s) + params | medium | can shard by bins | higher | depends on bin layout |
| PMM | inventory + oracle state + params | low/medium | depends on design | medium | quote updates may dominate |
| CLOB | market state + order matching state | varies | depends on matching engine design | high | crankless helps UX |
| TWAMM | long-term order state + execution legs | n/a | time-sliced | medium/high | often pairs with CLOB/AMM legs |
| Design | Parameters you can’t ignore | Tuning difficulty | Common footguns |
|---|---|---|---|
| CPAMM | fee bps, min liquidity lock, rounding rules | low | overflow in x*y, wrong deposit proportionality |
| StableSwap | amplification A, fee(s), admin fees, ramping | medium/high | bad A → fragility near peg/off-peg |
| CLMM | tick spacing, fee tier(s), init price, range UX | high | tick array provisioning, out-of-range UX |
| DLMM | bin step, dynamic fee curve, rebalancing rules | medium/high | bin skew → bad execution; edge-bin depletion |
| PMM | oracle choice, spread model, inventory/risk limits | very high | stale oracle, model blowups, adversarial flow |
| CLOB | tick size, lot size, maker/taker fees, risk limits | high | dust orders, spam, maker incentives |
| Bonding curve | virtual reserves, slope, caps, migration rules | high | sniping, MEV extraction, mispriced curve |
Token-2022 extensions change what “amount in” means.
| Token feature | What breaks in naive AMMs | Safe pattern | Designs most sensitive |
|---|---|---|---|
| Transfer fee | amount_in ≠ vault delta | compute dx = vault_after - vault_before | all curve AMMs |
| Transfer hook | extra logic executed on transfer | strict account lists; avoid re-entrancy assumptions | all; especially CPI-heavy |
| Confidential transfers | you can’t observe amounts easily | often incompatible without special support | most AMMs |
| Interest-bearing | balances drift over time | use observed balances; avoid cached reserves | all pool AMMs |
| Memo/metadata ext | usually fine | no-op | none |
Rule of thumb: if you don’t base math on observed vault deltas, you’re designing for 2019 SPL Token semantics.
| Design | Sandwich susceptibility | “Pick-off” risk | Mitigations that actually work | Notes |
|---|---|---|---|---|
| CPAMM | high | high | private routing, tighter fees, better routing, smaller hops | passive curve is easy to arb |
| StableSwap | medium | medium | similar; parameter robustness | off-peg events get brutal |
| CLMM | medium | high (LPs) | managed LP vaults; dynamic fees | LPs can get wrecked by volatility |
| DLMM | medium | medium/high | dynamic fees, bin strategy | depends on fee model |
| PMM | low/medium | medium | oracle + inventory + RFQ-style routing | “MM-like” behavior |
| CLOB | medium | medium | maker protections, anti-spam, risk controls | depends on market design |
| Bonding curve | very high | very high | anti-bot design + fair launch mechanics | launch is an MEV magnet |
| If your goal is… | Pick | Because | But be honest about… |
|---|---|---|---|
| ship fastest + minimal state | CPAMM | simplest accounts & math | contention + worse execution unless TVL is high |
| best majors execution with public LPs | CLMM | capital efficiency near spot | position UX + account explosion |
| stable pairs / correlated assets | StableSwap | low slippage near peg | parameter tuning & off-peg behavior |
| strategy-friendly liquidity | DLMM | bins + dynamic fees can match volatility | bin UX + more moving parts |
| tight quotes controlled by MM logic | PMM | can beat passive curves | oracle/model risk is the product |
| limit orders + pro features | CLOB | explicit bids/asks | maker bootstrapping + ops complexity |
| reduce impact of whale flow | TWAMM (+ a venue) | time-slicing | needs execution infra |
| token launch discovery path | Bonding curve → migrate | deterministic launch → deep liquidity later | launch MEV + migration design |
| Phase | Typical mechanism | Why it fits | What you usually add next |
|---|---|---|---|
| launch / discovery | bonding curve / small CP pool | simple, deterministic | anti-bot + migration |
| early liquidity | CPAMM | easy integrations | multiple fee tiers / incentives |
| scaling majors | CLMM or DLMM | better execution | managed LP vaults |
| pro trading | CLOB | limit orders | cross-margin/perps |
| flow optimization | PMM / RFQ-like | best execution for routed flow | private routing + inventory mgmt |
| large order UX | TWAMM | reduces impact | bundle/atomic strategies |
If you want users to deposit proportionally, you preserve:
amount_a / amount_b ≈ reserve_a / reserve_bA clamp-style approach:
and then you clamp the other side if user supplies less.
sqrt(Δa·Δb) is bootstrap-onlyFor subsequent deposits, use proportional minting:
Otherwise LP shares drift and you can mint unfairly.
A·B and must use u128If you verify k, do:
new_x * new_y >= old_k (often allowing rounding to favor LPs)u128 intermediates.amount_inFor fee-on-transfer tokens:
dx is vault_after - vault_before./// Compute CPAMM output (dy) from reserves (x, y) and effective input (dx_eff),
/// using u128 intermediates to avoid u64 overflow.
///
/// IMPORTANT (Token-2022):
/// - If the token can take a transfer fee, compute dx_eff from observed vault delta:
/// dx_eff = vault_x_after - vault_x_before
pub fn cpamm_out_amount(x: u64, y: u64, dx_eff: u64) -> u64 {
let x = x as u128;
let y = y as u128;
let dx = dx_eff as u128;
// dy = (y * dx) / (x + dx)
let den = x + dx;
if den == 0 {
return 0;
}
let dy = (y * dx) / den;
dy.min(u64::MAX as u128) as u64
}
| Design | “Simple swap” interface | Quote interface | Common integration shape | Gotcha |
|---|---|---|---|---|
| CPAMM | swap(amount_in, min_out) | deterministic from reserves | direct CPI | need observed deltas for Token-2022 |
| CLMM | same, but more accounts | tick-dependent | SDK computes accounts | account list errors are common |
| DLMM | similar | bin-dependent + dynamic fee | SDK required | bin selection correctness matters |
| PMM | often RFQ-like | oracle + MM params | router integration is key | “quote freshness” is the product |
| CLOB | order placement | book data | off-chain client + on-chain settle | maker ops are non-trivial |
| Design | Invariants to test | Edge cases | Suggested approach |
|---|---|---|---|
| CPAMM | k non-decrease (fee), no negative reserves | rounding, overflow, zero-liquidity | property tests with random swaps |
| StableSwap | monotonicity near peg, conservation | extreme imbalance, A ramps | fuzz + numerical bounds |
| CLMM | tick crossing correctness, fee growth | boundary ticks, out-of-range | differential tests vs reference |
| DLMM | bin transitions, dynamic fee function | bin depletion, fee spikes | fuzz + scenario sims |
| PMM | oracle staleness handling, risk limits | oracle outages, adversarial flow | simulation + kill-switch tests |
| CLOB | matching engine correctness | self-trade, partial fills | deterministic replay tests |
Uniswap v2 constant product (conceptual): https://docs.uniswap.org/contracts/v2/concepts/protocol-overview/how-uniswap-works
SPL Token Swap constant product curve (Solana reference implementation notes): https://github.com/solana-labs/solana-program-library/blob/master/docs/src/token-swap.md
Curve StableSwap overview (hybrid curve intuition): https://docs.curve.finance/stableswap-exchange/overview/
Orca Whirlpools (CLMM/CLAMM on Solana): https://dev.orca.so/ https://github.com/orca-so/whirlpools
Raydium constant product pool creation (CP/CPMM program docs): https://docs.raydium.io/raydium/pool-creation/creating-a-constant-product-pool https://docs.raydium.io/raydium/protocol/developers/addresses
Meteora (DLMM + enhanced constant product families): https://docs.meteora.ag/developer-guide/home https://docs.meteora.ag/user-guide/guides/how-to-use-dlmm https://docs.meteora.ag/overview/products/damm-v2/what-is-damm-v2 https://docs.meteora.ag/overview/products/dbc/what-is-dbc https://github.com/MeteoraAg/dynamic-bonding-curve
Lifinity (PMM / oracle-anchored market making on Solana): https://docs.lifinity.io/
Phoenix (crankless on-chain order book infra): https://github.com/Ellipsis-Labs/phoenix-v1
OpenBook v2 (CLOB program): https://github.com/openbook-dex/openbook-v2
Solana TWAMM: https://github.com/solana-labs/twamm
Token-2022 extensions (transfer fees / transfer hooks): https://solana.com/docs/tokens/extensions https://solana.com/docs/tokens/extensions/transfer-fees https://solana.com/developers/guides/token-extensions/transfer-hook
Jupiter routing (what venues get routed): https://support.jup.ag/hc/en-us/articles/22627174618780-Which-AMMs-does-Jupiter-route-through
tokio::select!).tokio-stream is an optional companion that:
Streams (e.g., ReceiverStream, BroadcastStream, IntervalStream);map, filter, merge, timeout, throttle, chunks_timeout, StreamMap) for declarative event pipelines.recv().await and coordinates with select!, you usually don’t need tokio-stream.#[tokio::main], tokio::spawn, tokio::select!tokio::sync::{mpsc, broadcast, watch, oneshot}tokio::time::{sleep, interval, timeout}tokio::signalrecv().await inside a select! loop.tokio-stream (adapters + combinators)Stream):
wrappers::ReceiverStream<T> ← mpsc::Receiver<T>wrappers::UnboundedReceiverStream<T>wrappers::BroadcastStream<T> ← broadcast::Receiver<T>wrappers::WatchStream<T> ← watch::Receiver<T>wrappers::IntervalStream ← tokio::time::IntervalStreamExt: next, map, filter, merge (with SelectAll), StreamMap (keyed fan-in), and time-aware ops (timeout, throttle, chunks_timeout) when the crate’s time feature is enabled.select! (“manual pump”) — no tokio-stream neededuse tokio::{select, signal, sync::mpsc};
#[tokio::main]
async fn main() -> anyhow::Result<()> {
let (tx, mut rx) = mpsc::channel::<String>(1024);
// Example producer
tokio::spawn(async move {
let _ = tx.send("hello".to_string()).await;
});
let mut sigint = signal::ctrl_c();
loop {
select! {
maybe = rx.recv() => {
match maybe {
Some(msg) => { tracing::info!("msg: {msg}"); }
None => break, // channel closed
}
}
_ = &mut sigint => {
tracing::info!("shutting down");
break;
}
else => break,
}
}
Ok(())
}
Pros
Cons
tokio-stream recommendeduse std::time::Duration;
use tokio::sync::mpsc;
use tokio_stream::{
wrappers::{ReceiverStream, IntervalStream},
StreamExt, // for .next() and combinators
};
enum AdapterEvent { User(String), Order(String), Heartbeat }
#[tokio::main]
async fn main() -> anyhow::Result<()> {
let (tx_user, rx_user) = mpsc::channel::<String>(1024);
let (tx_order, rx_order) = mpsc::channel::<String>(1024);
// Example producers
tokio::spawn(async move { let _ = tx_user.send("u1".into()).await; });
tokio::spawn(async move { let _ = tx_order.send("o1".into()).await; });
let ticker = tokio::time::interval(Duration::from_secs(1));
let users = ReceiverStream::new(rx_user).map(AdapterEvent::User);
let orders = ReceiverStream::new(rx_order).map(AdapterEvent::Order);
let beats = IntervalStream::new(ticker).map(|_| AdapterEvent::Heartbeat);
// Compose: merge multiple sources and shape the flow
let mut events =
users.merge(orders)
.merge(beats)
.throttle(Duration::from_millis(20));
while let Some(ev) = events.next().await {
match ev {
AdapterEvent::User(v) => tracing::info!("user: {v}"),
AdapterEvent::Order(v) => tracing::info!("order: {v}"),
AdapterEvent::Heartbeat => tracing::debug!("tick"),
}
}
Ok(())
}
Pros
impl Stream<Item = Event> to consumers.Cons
Receiver.| Aspect | Channel + tokio::select! (no tokio-stream) | Stream-first (uses tokio-stream) | What the dependency implies |
|---|---|---|---|
| Why it’s used | Pull from channels via recv().await, coordinate with select!. | Wrap Tokio primitives as Streams and/or use combinators. | Presence of tokio-stream signals a stream-centric composition. |
| Primary abstraction | Futures + channels + select!. | Stream<Item = T> + wrappers + StreamExt. | Stream API → extra crate. |
| Typical code | while let Some(x) = rx.recv().await {}, select! { ... } | ReceiverStream::new(rx).map(...).merge(...).next().await | Wrappers/combinators imply tokio-stream. |
| Fan-in / merging | Manual select! arms; verbose for many/dynamic sources. | merge, SelectAll, or StreamMap for succinct fan-in. | tokio-stream buys tools for multiplexing. |
| Timers / heartbeats | interval() polled in loops. | IntervalStream + timeout/throttle/chunks_timeout. | Time-aware ops rely on tokio-stream + features. |
| Public API shape | Pull: async fn next_event() -> Option<T>. | Stream: fn into_stream(self) -> impl Stream<Item = T>. | Exposing a stream often requires the crate. |
| Composability | Hand-rolled transforms. | One-liners with StreamExt (map/filter/batch). | Enables declarative pipelines. |
| Backpressure | Channel capacity governs it; explicit. | Same channels underneath; wrappers don’t change capacity. | Neutral; it’s about ergonomics. |
| Fairness/ordering | select! randomizes fairness per iteration. | Per-stream order preserved; cross-stream order depends on combinator. | Document semantics either way. |
| Testability | Manual harnesses around loops. | .take(n), .collect::<Vec<_>>(), etc. | Stream APIs are often easier to test. |
| Cost / deps | Lean; no extra crate. | Adds tokio-stream; thin adapter overhead. | Main cost is dependency surface. |
tokio-stream)next_event().tokio::time::timeout for per-item deadlines.use std::time::Duration;
use tokio::{sync::mpsc, time::timeout};
pub async fn pump_with_timeout(mut rx: mpsc::Receiver<String>) -> anyhow::Result<()> {
loop {
match timeout(Duration::from_secs(5), rx.recv()).await {
Ok(Some(msg)) => tracing::info!("msg: {msg}"),
Ok(None) => break, // channel closed
Err(_) => tracing::warn!("no event within 5s"),
}
}
Ok(())
}
Cargo.toml
[features]
default = []
stream-api = ["tokio-stream"]
[dependencies]
tokio = { version = "1", features = ["rt-multi-thread","macros","sync","time","signal"] }
tokio-stream = { version = "0.1", optional = true }
Client
#[cfg(feature = "stream-api")]
use tokio_stream::wrappers::ReceiverStream;
pub struct Client {
rx_inbound: tokio::sync::mpsc::Receiver<MyEvent>,
}
impl Client {
pub async fn next_event(&mut self) -> Option<MyEvent> {
self.rx_inbound.recv().await
}
#[cfg(feature = "stream-api")]
pub fn into_stream(self) -> ReceiverStream<MyEvent> {
ReceiverStream::new(self.rx_inbound)
}
}
broadcast so multiple consumers can subscribe.use tokio::sync::{mpsc, broadcast};
use tokio_stream::wrappers::BroadcastStream;
pub struct Client {
rx_inbound: mpsc::Receiver<Event>, // pull path
bus: broadcast::Sender<Event>, // stream path
_reader: tokio::task::JoinHandle<()>,
}
impl Client {
pub async fn next_event(&mut self) -> Option<Event> {
self.rx_inbound.recv().await
}
pub fn event_stream(&self) -> BroadcastStream<Event> {
BroadcastStream::new(self.bus.subscribe())
}
}
Stream without tokio-streamStream directly over mpsc::Receiver via poll_recv.use futures_core::Stream;
use pin_project_lite::pin_project;
use std::{pin::Pin, task::{Context, Poll}};
use tokio::sync::mpsc;
pin_project! {
pub struct EventStream<T> {
#[pin]
rx: mpsc::Receiver<T>,
}
}
impl<T> EventStream<T> {
pub fn new(rx: mpsc::Receiver<T>) -> Self { Self { rx } }
}
impl<T> Stream for EventStream<T> {
type Item = T;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.project().rx.poll_recv(cx)
}
}
ReceiverStream is a thin adapter; hot-path costs are typically parsing/allocations, not the wrapper.tokio::select! randomizes branch polling; stream fan-in fairness depends on the specific combinator (merge, SelectAll, StreamMap).impl Stream<Item = Event> or use stream combinators? → Use tokio-stream.recv().await and select!? → Tokio alone is fine.stream-api).tokio::select! macro:
https://docs.rs/tokio/latest/tokio/macro.select.htmltokio::sync::mpsc::Receiver:
https://docs.rs/tokio/latest/tokio/sync/mpsc/struct.Receiver.htmltokio-stream (overview):
https://docs.rs/tokio-streamReceiverStream:
https://docs.rs/tokio-stream/latest/tokio_stream/wrappers/struct.ReceiverStream.htmlStreamMap:
https://docs.rs/tokio-stream/latest/tokio_stream/struct.StreamMap.htmlStream trait:
https://docs.rs/futures-core/latest/futures_core/stream/trait.Stream.htmltokio-tungstenite WebSocketStream (Stream + Sink):
https://docs.rs/tokio-tungstenite/latest/tokio_tungstenite/struct.WebSocketStream.html