V.5 The distributed service "shop"

shop/
  oxl.toml  oxl.lock  oxl.schema.lock
  proto/inventory.proto  proto/payments.proto  proto/recs.proto
  src/
    main.oxl            // starting background tasks
    cluster.oxl         // the cluster and dependencies
    limits/limits.oxl   // constants
    config/config.oxl   // static and dynamic config
    model/model.oxl     // common types
    users/users.oxl
    orders/orders.oxl
    cart/cart.oxl
    checkout/checkout.oxl
    photos/photos.oxl
    search/search.oxl   // search with the A/B experiment "search v2"
  tests/
  traffic/search.oxlrec // recorded search traffic (`oxl trace --record`)

// All timeouts, limits and sizes are here and only here (part IV.1).
// Constants are evaluated at compile time and are suitable for attribute arguments.

// Timeouts
pub const STD_TIMEOUT       = 3s      // the default for all rpcs (cluster.defaults)
pub const CHECKOUT_DEADLINE = 10s     // the whole checkout, including the saga
pub const DRAIN_TIMEOUT     = 30s     // how long a node finishes requests when stopping
pub const SEARCH_P95_LIMIT  = 50ms    // guard of the A/B experiment "search v2": p95 of the candidate's search

// Limits and sizes
pub const PAGE_SIZE: u32    = 50      // the page size of lists
pub const DEFAULT_MAX_CART: u32 = 100 // the default value; changed on the fly through Features
pub const STATS_BUFFER: usize = 16    // the buffer of the statistics broadcast hub per subscriber

// Lifetimes and periods
pub const CART_IDLE         = 30m     // after how long an inactive cart is unloaded from memory
pub const STATS_PERIOD      = 5s      // how often the statistics are recomputed
pub const ORDER_EVENTS_RETENTION = 7d // how long order events are kept in the topic

// Sharding (change only with a data migration)
pub const ORDER_SHARDS: u32   = 64
pub const STORAGE_SHARDS: u32 = 256

// Other
pub const THUMB_SIZES: [u32; 2] = [256, 1024]   // photo thumbnail sizes, px
pub const AUTH_ISSUER       = "https://id.shop.example"
pub const PUBLIC_DOMAIN     = "shop.example"

use self.limits { DEFAULT_MAX_CART }

// The static config: read at start-up, does not change without a restart.
// Values are taken from flags, env (APP_DB_URL, APP_PAYMENTS_KEY…), a file or a profile.
pub config App {
  dbUrl: str = "builtin://shop"       // by default the cluster's built-in storage
  inventoryAddr: str                  // required: without a value the node does not start (code 78)
  paymentsAddr: str
  recsAddr: str = "localhost:9103"      // the recommendations model
  paymentsKey: secret<str>            // a secret: does not get into logs and traces
  profile dev {                       // values for local development
    inventoryAddr = "localhost:9101"
    paymentsAddr  = "localhost:9102"
    paymentsKey   = "dev-key"
  }
}

// The dynamic config: stored in the cluster, changes without a restart
// (oxl config set Features.maxCart 200).
pub dynamic Features {
  maxCart: u32 = DEFAULT_MAX_CART     // the limit of items in a cart
  liveStats: Flag = Flag.on           // the flag can be switched on for a percentage of users
}

// Identifiers are separate types: a UserId cannot be passed instead of an OrderId by accident
pub(api) type UserId  = distinct u64
pub(api) type OrderId = distinct u64
pub(api) type PhotoId = distinct u64

// Refined types are checked when a request is decoded —
// the handler will never see an invalid SKU or email
pub(api) type Sku     = str where matches(re"[A-Z0-9-]{3,32}")
pub(api) type Name    = str where len(1..=64)
pub(api) type Email   = str where matches(re".+@.+\..+")

// User roles; taken from the JWT claim "roles" (see cluster.oxl)
pub(api) enum AppRole { Customer, Support, Admin }

@derive(Clone, Eq, Show)
pub(api) data Item {
  sku: Sku
  title: str
  price: Money                        // price per piece
  qty: u32 where 1..=99               // quantity, checked on creation
}

// A snapshot of the cart at checkout time; passed to the saga
pub(api) data CartSnapshot { userId: UserId; items: List<Item>; total: Money }

// @open: new statuses may appear in the future, clients are required to handle "_"
@open pub(api) enum Status { New = 1, Paid = 2, Shipped = 3, Cancelled = 4 }

// Domain errors. Over the network they are passed to the client as typed gRPC details.
@error pub(api) data NotFound { what: str; id: u64 }
@error pub(api) data EmptyCart {}
@error pub(api) data CartFull { max: u32 }
@error pub(api) data EmailTaken {}
@error pub(api) data BadImage { reason: str }

use super.model.*

// An ORM table: the schema and migrations are derived from this declaration
@table("users") @index(.email)
pub(api) data User {
  @pk id: UserId                      // primary key
  name: Name
  email: Email
  createdAt: Time = now()             // the default value is evaluated on creation
}

// The service contract: the server, clients and .proto are generated from it
@v(1) pub(api) service Users {
  @readonly rpc get(id: UserId) -> User | NotFound            // readonly: safe to retry and cache
  @idempotent rpc register(name: Name, email: Email) -> User | EmailTaken   // a repeat with the same key returns the first response
}

impl Users {
  dep db: Db                          // a dependency, given by provide in cluster.oxl

  // Parameter types are taken from the service contract
  rpc get(id) { db.find(User, id) ?? fail NotFound{ what: "user", id: id.raw } }

  rpc register(name, email) {
    // The query is compiled to SQL at build time
    let taken = from u in db.User where u.email == email take 1
    if !taken.isEmpty() { fail EmailTaken{} }
    let u = User{ id: UserId(cluster.ids.next()), name, email }   // a cluster-unique id
    db.insert(u)
    u
  }
}

// Access for external calls: everything is denied by default
policy Users {
  allow get(id)  if caller.isUser(id) || caller.has(role: .Support)   // one's own profile or support
  allow register if caller is .Anonymous                             // registration — without logging in
}

use super.model.*
use super.limits { PAGE_SIZE, ORDER_SHARDS, ORDER_EVENTS_RETENTION, STATS_BUFFER }

// @owner: a user sees only their own orders, the condition is added to queries automatically.
// @shard: the table is split by userId — all orders of a user are in one shard.
@table("orders") @owner(.userId, bypass: [.Support, .Admin])
@index(.userId, .createdAt) @shard(by: .userId, count: ORDER_SHARDS)
pub(api) data Order {
  @pk id: OrderId
  userId: UserId
  items: List<Item>                   // stored serialised in one column
  total: Money
  receiptId: str                      // the payment id in the external system
  status: Status = Status.New
  createdAt: Time = now()
}

pub(api) enum OrderEvent {
  Created(id: OrderId, userId: UserId)
  StatusChanged(id: OrderId, userId: UserId, status: Status)
}

// A topic: every event is received by all subscribers; order is guaranteed within a userId
@retain(ORDER_EVENTS_RETENTION)
pub(api) topic OrderEvents: OrderEvent sharded by .userId

@derive(Show) pub(api) data Stats { ordersToday: u64; revenueToday: Money }

// The statistics topic keeps only the last value
@retain(count: 1) pub topic StatsUpdates: Stats

// A local broadcast hub on every api node: one stream from the topic is distributed to all clients of the node.
// A slow client loses old values (DropOldest), a new one immediately gets the last (replay: 1).
pub static LIVE_STATS: Broadcast<Stats> = Broadcast<Stats>(buffer: STATS_BUFFER, slow: .DropOldest, replay: 1)

@v(1) pub(api) service Orders {
  @readonly rpc recent(page: u32) -> List<Order>
  @readonly rpc get(id: OrderId) -> Order | NotFound
  // internal: called only by the saga, inaccessible to external clients
  @idempotent @internal rpc create(snap: CartSnapshot, receiptId: str) -> Order
  // resumable: on a disconnect the client reconnects and continues from the last offset
  @resumable rpc watch() -> stream<OrderEvent>
  rpc liveStats() -> stream<Shared<Stats>, Lagged>
}

impl Orders {
  dep db: Db

  rpc recent(page) {
    let uid = ctx.userId()?           // AccessDenied if the caller is not a user
    // the shard key (userId) is in the condition — the query goes to one shard
    from o in db.Order where o.userId == uid
      order by o.createdAt desc skip page * PAGE_SIZE take PAGE_SIZE
  }

  rpc get(id) { db.find(Order, id) ?? fail NotFound{ what: "order", id: id.raw } }

  rpc create(snap, receiptId) {
    let o = Order{ id: OrderId(cluster.ids.next()), userId: snap.userId,
                   items: snap.items, total: snap.total, receiptId }
    // A DB transaction: the order and the event are written together;
    // the event goes to the topic only after commit (outbox)
    atomic(db) {
      db.insert(o)
      emit OrderEvents(OrderEvent.Created(o.id, o.userId))
    }
    o
  }

  // Subscription only to the events of one's own user, starting with new ones
  rpc watch() { OrderEvents.subscribe(key: ctx.userId()?, from: .Latest) }

  // Every client is a new subscriber of the local hub
  rpc liveStats() { LIVE_STATS.subscribe() }
}

policy Orders {
  allow recent, get, watch if caller is .User           // any logged-in user (their own data — through @owner)
  allow liveStats if caller.has(role: .Support) || caller.has(role: .Admin)
  // create is not mentioned: it is inaccessible to external callers, allowed to the saga automatically by the call graph
}

// An ordinary function of the module; db is passed explicitly as a borrow
pub fn computeStats(db: &Db) -> Stats  db, time {
  let since = today().atStart().in(node.tz).time   // the start of today in the node's time zone
  let todays = from o in db.Order where o.createdAt >= since
  Stats{ ordersToday: todays.len() as u64,
         revenueToday: todays.fold(Money.ZERO, (a, o) => a + o.total) }
}

use super.model.*
use super.config { Features }
use super.limits { CART_IDLE }

// A sharded actor: one cart per user in the whole cluster.
// Lives on the node that owns the userId shard; the state (var) is saved to the journal after every message.
@idle(CART_IDLE)                      // an inactive cart is unloaded from memory but not lost
pub actor Cart sharded by userId: UserId {
  var items: List<Item> = []

  // Messages are processed strictly one at a time — no races over items
  msg add(item: Item) -> () | CartFull {
    let max = Features.get().maxCart.get()          // the current limit from the dynamic config
    if items.len() as u32 >= max { fail CartFull{ max } }
    items.push(item)
  }

  msg remove(sku: Sku) { items.retain(i => i.sku != sku) }

  // A copy of the contents for the saga: the cart itself stays with the actor
  msg snapshot() -> CartSnapshot {
    CartSnapshot{ userId, items: items.clone(),
                  total: items.fold(Money.ZERO, (a, i) => a + i.price * (i.qty as i64)) }
  }

  // Removes only the first n items — those that went into the order snapshot; items added after the snapshot remain
  msg removeOrdered(n: usize) { items = items.iter().skip(n).collect() }
}

// A user may work only with their own cart; userId is the actor's key
policy Cart { allow * if caller.isUser(userId) }

use super.model.*
use super.orders { Orders, Order }
use super.cart { Cart }
use super.limits { CHECKOUT_DEADLINE }
use super.photos { uploadPhoto as photoPipeline }   // renaming on import, so as not to confuse it with the rpc

// External systems are described in .proto; extern service and types are generated from them.
// reserve/release/charge/refund are marked idempotent in the proto — the saga needs that.
import proto "proto/inventory.proto"
import proto "proto/payments.proto"
import proto "proto/recs.proto"              // the recommendations model: rpc learn(user, items), does not affect carts and orders

// Clients of the external services; created via provide in cluster.oxl only on the saga role
dep inventory: Client<Inventory>
dep payments: Client<Payments>
dep recs: Client<Recs>

// A saga: steps go in turn; if a step fails, those already executed are rolled back
// by compensations (undo) in reverse order. Progress is written to the journal,
// so after a node crash another node of the saga role continues the saga.
saga placeOrder(snap: CartSnapshot) -> Order {
  inventory.reserve(items: snap.items)                              // 1. reserve the goods
      undo r => inventory.release(r.reservationId)                 //    rollback: release the reservation
  -> payments.charge(user: snap.userId.raw, amount: snap.total)   // 2. charge the money
      undo p => payments.refund(p.receiptId)                       //    rollback: return the money
  -> Orders.create(snap, it.receiptId)                             // 3. create the order; it is the result of step 2
}

// The public facade for clients
@v(1) pub(api) service Shop {
  rpc addToCart(item: Item) -> () | CartFull
  rpc checkout() -> Order | EmptyCart | OutOfStock | Declined | CompensationFailed
  rpc uploadPhoto(img: bytes) -> PhotoId | BadImage
  rpc search(q: str) -> List<str>                              // the search variant is chosen by the A/B experiment
}

impl Shop {
  // Cart(uid) is a reference to an actor; the call goes to the node where the cart lives
  rpc addToCart(item) { Cart(ctx.userId()?).add(item) }

  rpc checkout() {
    let uid = ctx.userId()?
    let res = await Cart(uid).snapshot()                        // take a snapshot of the cart
      -> s => if s.items.isEmpty() { EmptyCart{} } else { s }   // an empty cart is an error by value, the chain is interrupted
      -> placeOrder(it)                                         // start the saga
      within CHECKOUT_DEADLINE                                  // a common checkout deadline
    let order = res?
    // Cleanup must not be sent to the background: it could arrive after the user's next request
    // and wipe what was just added. So we wait; only the ordered items are removed.
    await Cart(uid).removeOrdered(order.items.len())
    // Only what does not affect the request state goes to the background: the order data for retraining the recommendations
    await order ~> recs.learn(user: uid.raw, items: it.items)
  }

  // The pipeline responds right after the photo is saved, processing continues in the background
  rpc uploadPhoto(img) => photoPipeline(ctx.userId()?, img)

  rpc search(q) => searchCatalog(q)                            // search/search.oxl
}

policy Shop { allow * if caller is .User }   // all methods — for logged-in users

use std.ab { ab }
use std.obs { metric }

data Entry { sku: str; title: str; tags: List<str> }

fn catalog() -> List<Entry> => [ … ]

// Variant a (control): by title only
pub fn searchByName(q: str) -> List<str> =>
  catalog().filter(e => e.title.contains(q.lower())).map(e => e.sku)

// Variant b (candidate): by title and by tags
pub fn searchByTag(q: str) -> List<str> =>
  catalog().filter(e => e.title.contains(q.lower()) || e.tags.contains(q.lower())).map(e => e.sku)

// Which variant a request gets is decided by the experiment (§18.5): in `oxl test` — by the user's hash on the recorded traffic,
// in a cluster — a dynamic Flag created by `oxl ab start "search v2"`
pub fn searchCatalog(q: str) -> List<str> {
  let found = match ab.variant("search v2") {
    "b" => searchByTag(q)
    _ => searchByName(q)
  }
  if !found.isEmpty() { metric.counter("search_found").inc() }   // the experiment's user metric
  found
}

use super.model.*
use super.limits { THUMB_SIZES }

dep blobs: BlobStore                  // file storage (provide in cluster.oxl)

// The "photo ready" topic: clients and the indexer subscribe to it
@retain(count: 1000) pub(api) topic PhotoEvents: PhotoReady
pub(api) data PhotoReady { id: PhotoId; owner: UserId; thumbs: List<str> }

data Photo { id: PhotoId; owner: UserId; image: Image }

// A pure function: checks the format and creates a Photo or returns BadImage
fn decodePhoto(owner: UserId, b: bytes) -> Photo | BadImage pure {
  let img = Image.decode(b) ?? fail BadImage{ reason: "unsupported format" }
  Photo{ id: PhotoId(cluster.ids.next()), owner, image: img }
}

// A pipeline: every step can run on its own node,
// data is passed from node to node directly, the client gets only respond.
pipeline uploadPhoto(owner: UserId, img: bytes) -> PhotoId {
  (o, b) => decodePhoto(o, b)                       // 1. on the current node; a bad file is an error to the client
  -> p => { blobs.put("orig/{p.id.raw}", p.image.encode()); p }   // 2. save the original
  -> respond it.id                                  // 3. the response to the client: the photo id; then the background
  -> id => makeThumbs(id, THUMB_SIZES)      on gpu  // 4. thumbnails on a GPU node; id is the value of respond
  -> moderate                              on ml  durable   // 5. moderation on an ML node; from this step progress is in the journal
  -> t => emit PhotoEvents(PhotoReady{ id: t.id, owner, thumbs: t.urls })   // 6. notify subscribers
} on_error(e, step) { log.error("photo pipeline failed at {step}: {e}") }   // errors after respond

use self.limits.*
use self.config { App }

// One binary for all pods; the pod's role determines what it brings up
cluster shop {
  seeds k8s(label: "app=shop")                      // how pods find each other
  auth { jwt(issuer: AUTH_ISSUER); roles from claim("roles") as AppRole }   // verification of user tokens
  defaults { timeout STD_TIMEOUT; drain DRAIN_TIMEOUT }   // defaults for all rpcs
  trace_sample 0.05                                 // the fraction of requests traced by default (the flag --trace-sample overrides)

  // api: public services and carts, outwards on 443 with an ACME certificate
  role api     { serves Users, Orders, Shop; hosts Cart; replicas 3..10; expose 443 tls(acme: PUBLIC_DOMAIN) }
  // storage: the built-in storage (journal, topics, actor state), 3 replicas of every shard
  role storage { journal; shards STORAGE_SHARDS; replication 3 }
  // saga: executes sagas and the statistics leader task
  role saga    { runs sagas; runs leader("stats") }
  // gpu and ml: only for the pipeline steps on gpu / on ml; pods with the needed labels
  role gpu     { needs label("gpu=true") }
  role ml      { needs label("ml=true") }

  // The pod's role: environment variable → pod label → automatically by replica deficit
  assign by env("OXL_ROLE") else label("oxl/role") else auto
}

// Dependencies: created once per node at start-up.
// Without for role — on all roles; clients of external systems are needed only by the saga.
provide Db = Db.open(App.get().dbUrl)
provide BlobStore = BlobStore.open(App.get().dbUrl)
provide Client<Inventory> = Inventory.connect(App.get().inventoryAddr, tls: {})            for role saga
provide Client<Payments>  = Payments.connect(App.get().paymentsAddr, tls: {},
                                             auth: .Bearer(App.get().paymentsKey))         for role saga
provide Client<Recs>      = Recs.connect(App.get().recsAddr, tls: {})                      for role api

use self.orders { StatsUpdates, LIVE_STATS, computeStats }
use self.limits { STATS_PERIOD }

dep db: Db

// main runs on every node after the roles have started.
// The node does not stop after main as long as it hosts services and background tasks.
fn main() net, db, time {
  // Register the leader task: it runs on exactly one node with runs leader("stats")
  // (the saga role); if it fails — on another one. On the other nodes the call does nothing.
  cluster.leader("stats") {
    every(STATS_PERIOD) { emit StatsUpdates(computeStats(&db)) }   // publish to the topic every 5 s
  }

  // On every api node: one topic subscription → a local broadcast → all clients of the node.
  // Thus the cluster topic is read once per node, not once per client.
  if cluster.self.hasRole("api") {
    node.background.spawn {
      for await s in StatsUpdates.subscribe(from: .Latest) { LIVE_STATS.publish(s) }
    }
  }
}

// tests/money_test.oxl — an ordinary unit test
test "money arithmetic" {
  assert Money.rub(10) + Money.rub(5) == Money.rub(15)
  assert encode(Money.rub(1))! == encode(100i64)!      // over the network Money is an i64 in kopecks
}

// tests/checkout_test.oxl

// A fixture: the value before yield is given to the test, the code after yield is cleanup
fixture customer() -> UserId { yield UserId(42) }

// A unit test: an in-memory node with the role all, services and actors in-process
test "checkout empty cart" with customer as uid {
  asUser(uid) { assert_fails(Shop.checkout(), EmptyCart) }   // a call on behalf of the user
}

test "payment timeout is retried by saga" with customer as uid {
  // Mocks of external services: responses in call order
  mock Inventory { reserve(_) => ReserveResp{ reservationId: "r1" } always }
  mock Payments  { charge(_)  => fail Timeout once, then ChargeResp{ receiptId: "p1" } }
  asUser(uid) {
    Shop.addToCart(TEST_ITEM)?        // ? in a test: an error fails the test
    assert Shop.checkout() is Order
  }
  // The first charge failed by timeout, the second succeeded
  assert Payments.calls(.charge).count == 2
  // Both calls came with the same idempotency key — the money is not charged twice
  assert Payments.calls(.charge).all(c => c.ctx.idempotencyKey == Payments.calls(.charge)[0].ctx.idempotencyKey)
}

// Simulation: the network, clocks and scheduler are deterministic, the test is run for 300 seeds
@tag(slow)
test sim "checkout survives saga node crash" seeds 1..=300 with customer as uid {
  mock Inventory { reserve(_) => ReserveResp{ reservationId: "r1" } always; release(_) => Ok always }
  mock Payments  { charge(_)  => ChargeResp{ receiptId: "p1" } always; refund(_) => Ok always }
  net.latency(5ms..200ms)             // random network delays (depend on the seed)
  node.kill("saga-0") after 50ms      // kill the saga node in the middle of checkout
  let r = asUser(uid) { Shop.addToCart(TEST_ITEM) -> Shop.checkout() }
  assert r is Order | Timeout
  assert r is Order || Inventory.calls(.release).count >= 1     // either an order, or the reservation released
}

// An integration test: a real cluster of 3 nodes in the process
@tag(slow, integ)
test integ "order events reach subscriber" with cluster(nodes: 3), customer as uid {
  let sub = asUser(uid) { Orders.watch() }                    // subscribe before checkout
  asUser(uid) { Shop.addToCart(TEST_ITEM) -> Shop.checkout() }?
  assert (await sub.first()) is OrderEvent.Created            // the first event is the order creation
}

// An A/B test: the recorded search traffic is run on both variants, the metrics are compared with significance;
// a candidate that violates the guard fails the test (in a cluster — rolls the flag back)
test ab "search v2" {
  variant a = searchByName
  variant b = searchByTag
  split by .User 50/50
  metric search_found, latency.p95, error_rate
  guard latency.p95 < SEARCH_P95_LIMIT
  traffic recorded("traffic/search.oxlrec")
}

# Development
$ oxl dev --nodes 5                               # the whole cluster locally, roles are distributed auto

# Tests
$ oxl test                                         # by default: unit + integ + contract
$ oxl test --kind sim --seeds 1..1000 -f checkout  # only simulations with "checkout" in the name
$ oxl test --tag !slow --changed main              # fast tests affected by changes since main
$ oxl test --seed 217 --debug                      # reproduce a failed seed under the debugger
$ oxl test --kind ab                               # A/B tests on recorded traffic

# An A/B experiment in a cluster
$ oxl trace --record traffic/search.oxlrec --cluster shop.internal:7000 --for 10m --service shop.checkout.Shop
$ oxl ab start "search v2" --cluster shop.internal:7000   # creates a Flag and splits traffic
$ oxl ab status "search v2" --cluster shop.internal:7000  # variant metrics and significance

# Build and deploy
$ oxl build --profile prod --image registry.local/shop:0.3.0   # a static binary in an OCI image
$ kubectl apply -f k8s/                            # one image for all pods, the role is the label oxl/role

# Observation and debugging
$ ./shop --role api --inspect-ui                   # the state web interface at 127.0.0.1:8090
$ oxl inspect --cluster shop.internal:7000         # the same interface from the developer's machine
$ oxl debug enable --node api-2 --for 30m          # allow debugging of the node for 30 minutes
$ oxl debug --node api-2 --cluster shop.internal:7000   # connect through the cluster, without opening a port

$ oxl gen --py --out clients/py

# The client is generated by the command oxl gen --py: types and errors match the Shop contract
from shop_client import Shop, Item, EmptyCart

shop = Shop("shop.example:443", token=jwt)          # gRPC over TLS, the token is the user's JWT

# price is Money, over the network it is an i64 in kopecks (150000 = 1500 ₽)
shop.add_to_cart(Item(sku="BOOK-1", title="Book", price=150000, qty=1))

try:
    order = shop.checkout()
except EmptyCart:                                    # an OxL @error type is a separate Python exception
    ...