janus/rama

Clojure utilities on top of Rama: managed objects with dependency injection, batch segmacros, symbol helpers, temp-state utilities, and depot Datasets.

janus.rama.api — managed objects, batch segmacros, symbol / temp-state helpers.
janus.rama.api.ds — Dataset protocol surface: live depot extraction, on-disk dumps, and into-depot!.

Architecture, concepts, and ADRs for this component live next to the code under components/rama/doc/ (source of truth). This page is the short user-facing summary.

janus.rama.api should be required with :refer :all when using its Rama segmacros, because those are resolved by name at compile time.

Setup

Add the dependency:

;; deps.edn
{janus/rama {:mvn/version "..."}}

Require the API namespace(s):

(require '[janus.rama.api :refer :all])
;; Datasets / depot dump-load:
(require '[janus.rama.api.ds :as ds])

Managed Objects

The Problem

Rama provides in com.rpl.rama.integration means to define ManagedResources with semantics on how they are shared (per task thread, per worker, etc…).

This package provides a way to specify such managed resources in clojure. In addition, it allows that these ManagedResources themselves share other managed resources (e.g. API clients sharing a common http-client).

`per-worker`

Creates a ManagedResourceSpec — a serializable description of how to build a resource.

(per-worker key ctor & args)

key — a keyword uniquely identifying the resource (e.g. ::my-http-client).
ctor — a symbol or var pointing to the constructor function.
args — any arguments forwarded to the constructor via (apply ctor args).

The spec itself is Nippy-serializable and safe to embed in topology code and/or configuration.

`as-object`

Wraps a ManagedResourceSpec into a ManagedResourceWrapper that implements both TaskGlobalObject and IDeref. Declare it with declare-object and deref it at runtime to obtain the live resource.

(declare-object *my-client (as-object (per-worker ::my-client #'make-client opts)))

;; Inside topology code, deref to get the live resource:
@*my-client

`resource`

Extracts the live resource from a ManagedResource instance directly. Primarily used internally by the DI resolution machinery.

(resource managed-resource) ;; => the live resource value

`secret`

Convenience constructor for secrets as per-worker managed resources. Uses janus.secrets.api/get under the hood.

(secret id) ;; id is a qualified keyword

Example:

(declare-object *token (as-object (secret :cloudflare/api-token)))

;; In topology code:
@*token ;; => "sk-..."

Since secret returns a ManagedResourceSpec, it can be nested inside another per-worker spec — the DI machinery resolves it automatically:

(def ld-client
  (per-worker ::ld-client
    #'my.livingdocs/client
    {:base-uri "https://api.livingdocs.io"
     :token    (secret :livingdocs/api-token)}))

(declare-object *ld-client (as-object ld-client))

The secret is fetched once per worker and injected into the constructor map before my.livingdocs/client is called.

See janus/secrets for the full secrets documentation (lookup chain, naming conventions, CLI tool).

Shared Dependency Injection

The key feature of managed objects is automatic dependency resolution with identity sharing. If a spec’s args contain other ManagedResourceSpec instances, they are resolved before the outer constructor is called. Crucially, specs with the same key resolve to the same instance — enabling shared dependencies.

Example: Shared Base Resource

;; Spec A: a base resource
(def spec-a (per-worker ::a #'make-base-resource :config))

;; Specs B and C both depend on A
(def spec-b (per-worker ::b #'make-service-b spec-a))
(def spec-c (per-worker ::c #'make-service-c spec-a))

When B and C are materialised:

A is resolved first (because it appears in both args lists).
B receives the live instance of A as its first argument.
C receives the same identical instance of A — not a copy.

(identical? (get-a-from-b) (get-a-from-c)) ;; => true

This means multiple managed objects can safely share a single connection pool, cache, or any other stateful resource.

Automatic Cleanup of Dependencies

When a resource has nested dependencies (i.e. its args contained other specs that were resolved into ManagedResource instances), closing the outer resource automatically closes all of its sub-resources. You do not need to manage the lifecycle of inner dependencies manually — closing the top-level object is sufficient.

;; When *service-b is closed (e.g. at end of task), the inner
;; resource created from spec-a is also closed automatically.
(declare-object *service-b
  (as-object (per-worker ::b #'make-service-b
               (per-worker ::a #'make-connection-pool))))

Real-World Pattern

In practice, config maps pass specs that are resolved by topology setup code:

;; Config map defines how to build the API client
(def config {:api-client (per-worker ::api-client #'make-api-client opts)})

;; Topology setup declares the object
(let [api-client (derive-rama-var* '* name 'external-api-client)]
  (declare-object* setup api-client (as-object (:api-client config))))

;; In topology code, deref the object to call the API
(call-api-client api-client instruction)
;; where call-api-client derefs: @api-client

See io.forward-publishing.janus.topologies.polling and io.forward-publishing.janus.topologies.cache-purging for full examples.

Cyclic dependencies are not handled. If spec A depends on spec B and spec B depends on spec A, the resolution will loop indefinitely.

defbatch

A macro for writing topology-level batch segmacros with familiar Clojure syntax.

(defbatch name doc-string? meta? [inputs... :> outputs...] body...)

inputs — standard Clojure binding forms (symbols, destructuring).
:> — separates inputs from outputs (optional).
outputs — simple symbols for generated temp pstates.
body — Rama topology expressions.

The macro automatically sanitizes ramavar symbols (*foo, %bar, $$baz) so they don’t collide across invocations. Under the hood it expands into defbasicsegmacro.

Input-only (aggregation)

When there are no :> outputs, the batch typically aggregates into an existing pstate:

(defbatch <<sync-graph [%new-ids $$pstate]
  (%new-ids :> *new-id)
  (aggs/+set-agg $$pstate *new-id))

With outputs (temp pstate generation)

When :> is present, the batch produces temporary pstates:

(defbatch execution-plan [%new-ids :> $$plan]
  (%new-ids :> *id)
  (materialize> *id :> $$plan))

Other Utilities

Temp State

temp-state — a segmacro that creates a temporary nil pstate. Call at the top level of a microbatch topology.

(temp-state :> $$my-temp)

Ramaop Generators

These segmacros produce anonymous ramaops from materialised or temporary pstates:

Segmacro Description

Segmacro	Description
`materialized->ramaop`	`[n materialized :> op]` — emits contents of a materialized pstate across all tasks. `n` is the tuple size.
`this-task-materialized->ramaop`	Same as above but only on the current task.
`temp-pstate->ramaop`	`[temp-pstate path :> op]` — emits values from a temp pstate path across all tasks.
`this-task-temp-pstate->ramaop`	Same as above but only on the current task.

materialized->ramaop

[n materialized :> op] — emits contents of a materialized pstate across all tasks. n is the tuple size.

this-task-materialized->ramaop

Same as above but only on the current task.

temp-pstate->ramaop

[temp-pstate path :> op] — emits values from a temp pstate path across all tasks.

this-task-temp-pstate->ramaop

Same as above but only on the current task.

srange-dynamic Helpers

Function Description

Function	Description
`second>`	`(second> coll)` — returns `(min 1 (count coll))`. Useful as the second argument to `srange-dynamic`.
`count>`	`(count> coll _)` — returns `(count coll)`. Useful as the end argument to `srange-dynamic`.

second>

(second> coll) — returns (min 1 (count coll)). Useful as the second argument to srange-dynamic.

count>

(count> coll _) — returns (count coll). Useful as the end argument to srange-dynamic.

Symbol Helpers

Utilities for manipulating Rama variable symbols (*foo, %bar, $$baz):

Function Description

Function	Description
`sans-rama-prefix`	Strips the Rama prefix (`*`, `%`, `$$`) from a symbol name.
`derive-rama-var*`	`(derive-rama-var* prefix base)` or `(derive-rama-var* prefix base suffix)` — derives a new Rama variable symbol. E.g. (derive-rama-var* ' '*foo)` -> `'foo.
`derive-rama-var`	Macro version of `derive-rama-var*` — symbols need not be quoted.

sans-rama-prefix

Strips the Rama prefix (*, %, $$) from a symbol name.

derive-rama-var*

(derive-rama-var* prefix base) or (derive-rama-var* prefix base suffix) — derives a new Rama variable symbol. E.g. (derive-rama-var* ' '*foo)` -> `'foo.

derive-rama-var

Macro version of derive-rama-var* — symbols need not be quoted.

Depot Utilities & Datasets

Public namespace: janus.rama.api.ds (not re-exported from janus.rama.api).

A Dataset is a multi-partition stream of depot records (Dataset protocol). Live extraction, on-disk dumps, and ad-hoc fixtures all implement it. Sinks (write!, into-depot!) take any Dataset.

Live ranges are frozen at construction (point-in-time window, not a live tail). into-depot! uses foreign-append! — the target depot partitioner places records; source partition indices are not preserved.

How to dump a depot to a folder and load a dump into a depot: Depot dumps and loads.

(require '[janus.rama.api.ds :as ds])

(ds/write-depot! foreign-depot "/var/tmp/dump" :profile :archive)
(ds/into-depot! other-depot (ds/open "/var/tmp/dump"))

Profiles: :archive (default — Nippy + Brotli) and :fixture (EDN lines for git). On-disk layout: manifest.edn + partitions/<i>.part.

Architecture and ADRs: components/rama/doc/ (architecture/datasets.adoc, decisions.adoc).

API reference

Function Description

Function	Description
`depot-dataset`	`(depot-dataset depot)` or `(depot-dataset depot offset-back)` — Dataset over a foreign depot; optional trailing window per partition. `offset-back` of `0` is empty. Optional `:page-size` for `foreign-depot-read` paging (default 250).
`depot-sample`	`(depot-sample depot offset-back)` — Dataset for a fixed trailing window (same freeze path as `depot-dataset`). Supports legacy keyword lookup (`:offsets`, `:counts`, `:total-count`, …) via `ILookup`, but is not a real map (`map?`/`assoc`/`pr-str` of the sample itself). Use `serialize-sample` for EDN.
`depot-count`	`(depot-count depot)` or `(depot-count depot offset-back)` — counts items in a depot, optionally limited to the last `offset-back` items per partition. `offset-back` of `0` counts zero.
`depot-seq`	Seq convenience: `(depot-seq depot pkey)` or `(depot-seq depot pkey offset-back)` — lazy sequence for the partition matching `pkey` (queries only that partition). Prefer `depot-dataset` + `partition-seq` for new code.
`depot-sample-seq`	Seq convenience: `(depot-sample-seq sample)` interleaves partitions (legacy analytics order); `(depot-sample-seq sample :pkey k)` / `:partition-index i` for one partition. Prefer `ds-seq` / `partition-seq` for new code.
`serialize-sample`	`(serialize-sample sample)` — EDN-friendly descriptor (offsets, counts, …) without a live depot handle. Preferred path for printing/persisting samples.
`restore-sample`	`(restore-sample depot serialized)` — re-attaches a foreign depot to a serialized descriptor; returns a Dataset. Validates offset shape up front.
`num-partitions` / `partition-seq` / `info` / `ds-seq`	Protocol helpers over any Dataset (live, on-disk, or ad-hoc). `ds-seq` is partition-major.
`open` / `dataset` / `write!` / `write-depot!` / `into-depot!` / `read-manifest`	On-disk codec and sinks: open a dump directory, build an ad-hoc Dataset, persist any Dataset, sugar for live depot dumps, append Dataset records into a depot (default ack `:append-ack`).

depot-dataset

(depot-dataset depot) or (depot-dataset depot offset-back) — Dataset over a foreign depot; optional trailing window per partition. offset-back of 0 is empty. Optional :page-size for foreign-depot-read paging (default 250).

depot-sample

(depot-sample depot offset-back) — Dataset for a fixed trailing window (same freeze path as depot-dataset). Supports legacy keyword lookup (:offsets, :counts, :total-count, …) via ILookup, but is not a real map (map?/assoc/pr-str of the sample itself). Use serialize-sample for EDN.

depot-count

(depot-count depot) or (depot-count depot offset-back) — counts items in a depot, optionally limited to the last offset-back items per partition. offset-back of 0 counts zero.

depot-seq

Seq convenience: (depot-seq depot pkey) or (depot-seq depot pkey offset-back) — lazy sequence for the partition matching pkey (queries only that partition). Prefer depot-dataset + partition-seq for new code.

depot-sample-seq

Seq convenience: (depot-sample-seq sample) interleaves partitions (legacy analytics order); (depot-sample-seq sample :pkey k) / :partition-index i for one partition. Prefer ds-seq / partition-seq for new code.

serialize-sample

(serialize-sample sample) — EDN-friendly descriptor (offsets, counts, …) without a live depot handle. Preferred path for printing/persisting samples.

restore-sample

(restore-sample depot serialized) — re-attaches a foreign depot to a serialized descriptor; returns a Dataset. Validates offset shape up front.

num-partitions / partition-seq / info / ds-seq

Protocol helpers over any Dataset (live, on-disk, or ad-hoc). ds-seq is partition-major.

open / dataset / write! / write-depot! / into-depot! / read-manifest

On-disk codec and sinks: open a dump directory, build an ad-hoc Dataset, persist any Dataset, sugar for live depot dumps, append Dataset records into a depot (default ack :append-ack).