docs/guides/datatype-collections.md

DataType - Collection classes

UnsortedSet, Sorted Set, List, and Hash data types all include the Collection module. This guide covers two performance-sensitive concerns: writing many elements efficiently (a single bulk command instead of one round-trip per element), and iterating large collections efficiently via each and each_record.

Bulk writes — single round-trip mutations

Collection mutations are immediate — every call hits Valkey/Redis right away, unlike scalar field setters which are deferred until save. Each call also runs warn_if_dirty! and cascades expiration. (See the write-model notes in AGENTS.md for the deferred-vs-immediate split.)

Multi-element adds issue one command for the whole batch, not one per element. Populating a large collection is therefore a single round-trip even without an explicit pipeline.

The argument shape follows the collection's structure, and is consistent across the codebase:

Value-only collections (UnsortedSet, ListKey) take a variadic splat; arguments are flattened and nil-compacted.
Keyed/pair collections (HashKey is field => value, SortedSet is member => score) take a single Hash via update (aliased merge!), raising ArgumentError on a non-Hash.

Type	Bulk method	Call shape	Redis command
`UnsortedSet`	`add(*values)`	`tags.add(:a, :b, :c)`	one `SADD`
`ListKey`	`push(values)` / `unshift(values)`	`log.push(1, 2, 3)`	one `RPUSH` / `LPUSH`
`HashKey`	`update(hash)` / `merge!`	`cfg.update(a: 1, b: 2)`	one `HMSET`
`SortedSet`	`update(hash)` / `merge!`	`board.update("alice" => 1000, "bob" => 850)`	one `ZADD`

tags.add(:ruby, :redis, :valkey)              # 1 SADD, returns self
log.push("a", "b", "c")                        # 1 RPUSH → [a, b, c]
board.update("alice" => 1000, "bob" => 850)    # 1 ZADD, returns new-member count (2)
board.merge!("alice" => 1200)                  # 1 ZADD, score updated → returns 0

Behavior notes:

Ordering: push preserves argument order; unshift prepends each element in turn, so unshift(a, b, c) leaves the list head as c, b, a (Redis LPUSH semantics — unchanged from the prior per-element implementation). Sets are unordered; sorted sets order by score.
Empty input is a no-op: add() / push() / update({}) issue no command. Set/list adds return self; SortedSet#update returns 0.
SortedSet#add(val, score, …) is unchanged and not bulk — it takes a single member plus score and the conditional ZADD options (nx:, xx:, gt:, lt:, ch:). An Array passed as val is stored as one JSON-encoded member, not exploded into many. Use update/merge! for bulk insertion.

Capped collections — `max_length:`

SortedSet and ListKey accept a max_length: option that caps the collection at write time. Every member-creating write trims in the same operation, so the collection never stays over the cap:

# Standalone
events = Familia::SortedSet.new 'events', max_length: 100

# Horreum class declaration
class Customer < Familia::Horreum
  sorted_set :audit_events, max_length: 10_000
  list :recent_errors, max_length: 50
end

Every collection exposes the configured cap through #max_length, which returns nil when uncapped:

customer.audit_events.max_length  #=> 10_000
customer.tags.max_length          #=> nil

It is read-only: the cap is fixed at definition time so that validation can happen once, up front. To change it, redeclare the collection.

SortedSet: top-N by score

max_length: N retains the N highest-scoring members — the trim is ZREMRANGEBYRANK key 0 -(N+1). This is "newest N" only when scores are timestamps; with arbitrary scores it is simply top-N by score. Capping applies to all member-creating paths: add (and << / []=), update / merge!, and increment / decrement (ZINCRBY creates the member if absent). The trim runs unconditionally after the write, which is a cheap no-op when under the cap and harmless when a conditional nx:/xx:/gt:/lt: add skipped the write.

ListKey: per-end semantics

The cap keeps the elements nearest the end you wrote to — this asymmetry is intentional:

push (RPUSH) trims from the head, keeping the newest tail elements.
unshift (LPUSH) trims from the tail, keeping the newest head elements.

Atomicity

Standalone, each write+trim pair is wrapped in its own MULTI, so a crash cannot leave the collection over-cap; the write command's documented return value (e.g. add's Boolean, update's new-member count) is preserved. Inside a caller transaction (Fiber[:familia_transaction]) or pipeline, the pair is issued bare and the outer MULTI covers both — Redis MULTI does not nest.

What is not capped

Writes that bypass the instance's write methods are out of scope: unionstore / interstore / diffstore destination keys, RESTORE, and any external client writing the key directly.

ListKey also leaves its conditional and positional writers uncapped by design, since none of them is an append in the sense the per-end trim assumes:

pushx / unshiftx — conditional appends that no-op on a missing key.
insert (LINSERT) and set (LSET) — position-relative writes.
move (LMOVE) when this list is the destination; the cap belongs to the method being called, and move is called on the source.

Adding max_length: to a collection that is already over the cap does not trim it retroactively — nothing runs at definition time. The cap re-asserts itself on the next capped write; to enforce it sooner, call enforce_max_length!.

Enforcing the cap on existing data — `enforce_max_length!`

Adding max_length: to a live production collection is a no-op until something writes to it. enforce_max_length! applies the cap immediately — the migration step after declaring a cap on an existing collection — and returns the number of elements removed (0 when already within the cap). Calling it on an uncapped collection raises Familia::Problem rather than silently doing nothing.

customer.audit_events.enforce_max_length!   #=> 4215 (keeps the 10_000 highest scores)
customer.recent_errors.enforce_max_length!  #=> 12   (keeps the newest 50 at the tail)

SortedSet keeps the max_length highest-scoring members — the same ZREMRANGEBYRANK trim every capped write applies.
ListKey must be told which end survives, since the cap's per-end semantics belong to the write method: keep: :tail (the default) matches a push-fed list, keep: :head an unshift-fed one. LLEN and LTRIM run in one MULTI, so the removed count is exact even under concurrent writes.

Capping a `participates_in` collection

participates_in (and class_participates_in) accept max_length: directly. The collection declared on the target carries the cap, and record_class: is still threaded automatically so each_record keeps working:

class FeedItem < Familia::Horreum
  feature :relationships
  participates_in Owner, :activity, score: :created_at, max_length: 1000
end

The value is validated at class-definition time: it must be a positive Integer, and only :sorted_set and :list implement capping — max_length: with any other type: raises ArgumentError.

A capped participation collection is a recent-N view, not authoritative membership. The trim evicts members silently, without touching each participant's participations reverse-index set. Membership checks (collection.member?, the participant's in_*? methods) query the collection live and stay accurate after eviction, but current_participations reads the reverse index and can list collections the participant was trimmed out of. Destruction cleanup tolerates this — removing a member that was already evicted is a no-op — so the staleness over-reports but never breaks anything.

Pre-declaring the collection on the target still works — participation never overwrites an existing accessor — but then the two declarations must agree. A participates_in whose max_length: differs from the pre-declared cap (including an uncapped pre-declaration) raises ArgumentError at definition time rather than silently keeping the wrong cap; omitting max_length: from participates_in keeps whatever the pre-declaration says, capped or not. When you pre-declare, pass record_class: yourself, since you are replacing the declaration participation would otherwise have made — without it, each_record on the collection has nothing to hydrate:

class Owner < Familia::Horreum
  identifier_field :owner_id
  field :owner_id
  # Must come first — participates_in skips a collection that already exists
  sorted_set :activity, max_length: 1000, record_class: 'FeedItem'
end

class FeedItem < Familia::Horreum
  feature :relationships
  participates_in Owner, :activity, score: :created_at, max_length: 1000  # must match (or be omitted)
end

Validation and the old `:maxlength` spelling

max_length: must be a positive Integer — anything else raises ArgumentError at definition time (max_length: 0 would delete everything on every write).
Passing max_length: to a type that does not implement it (HashKey, UnsortedSet, StringKey, Counter, …) raises ArgumentError rather than being silently ignored.
The old :maxlength spelling was never honored and remains ignored; it now emits a warning ([familia] :maxlength is ignored; rename to max_length:). It is deliberately not treated as an alias — honoring it would start mass-deleting previously untrimmed data on upgrade. Rename to max_length: explicitly to opt in to trimming.

The iteration methods each and each_record efficiently handle large collections by paginating through Valkey/Redis data structures, but they serve different purposes and yield different results. Here's how the two iterate, using ModelClass.instances (a SortedSet with reference: true) as the running example.

`each` — yields members (identifiers, raw strings)

each is implemented per type. For the instances SortedSet, it pages through the ZSET with either ZRANGEBYSCORE (when since:/until: are given) or ZSCAN (unbounded), yielding one deserialized member at a time.

flowchart TD
  Caller["ModelClass.instances.each { |id| ... }"] --> EachImpl["SortedSet#each"]
  EachImpl --> Decide{since/until?}
  Decide -- yes --> ZRBS["ZRANGEBYSCORE key min max LIMIT 0 batch_size WITHSCORES"]
  Decide -- no --> ZSCAN["ZSCAN key cursor COUNT batch_size"]
  ZRBS --> Page["Page of raw members"]
  ZSCAN --> Page
  Page --> Yield["yield deserialize_value(member)"]
  Yield --> More{more pages?}
  More -- yes --> Decide
  More -- no --> Done["return self"]

Per-type variations:

ListKey#each — paginates with LRANGE start stop (no SCAN equivalent)
UnsortedSet#each / HashKey#each — SSCAN / HSCAN, optional matching: glob
SortedSet#each — ZRANGEBYSCORE (bounded) or ZSCAN (unbounded)

You get identifiers only. No record loading. One Redis round-trip per page.

`each_record` — yields loaded Horreum records

each_record is defined once in CollectionBase and delegates to each to collect identifiers, then batches them into record_class.load_multi (pipelined HGETALLs), filters ghosts, and yields the live records.

flowchart TD
  Caller["ModelClass.instances.each_record { |rec| ... }"] --> ER["each_record(batch_size, pipeline, **filters)"]
  ER --> Validate{"pipeline <= batch_size?"}
  Validate -- no --> Raise["raise ArgumentError"]
  Validate -- yes --> CallEach["each(**filters) do |member|"]
  CallEach --> Extract["id = member.is_a?(Array) ? member.last : member"]
  Extract --> Buffer["buffer << id"]
  Buffer --> Full{"buffer.size >= batch_size?"}
  Full -- no --> CallEach
  Full -- yes --> Load["record_class.load_multi(ids)  -- pipelined HGETALLs"]
  Load --> Compact["live = records.compact  -- drop ghosts"]
  Compact --> Mode{pipeline?}
  Mode -- nil --> Serial["live.each { |r| block.call(r) }"]
  Mode -- positive --> Pipe["live.each_slice(pipeline) do |group|<br/>record_class.pipelined { group.each &block }<br/>end"]
  Serial --> Clear["buffer.clear; resume each"]
  Pipe --> Clear
  Clear --> CallEach
  CallEach -. each exhausted .-> Flush["process_batch(buffer) if any remain"]
  Flush --> Return["return self"]

Concrete timeline for `User.instances.each_record(batch_size: 100, pipeline: 25) { |u| u.touch! }`

SortedSet#each (ZSCAN page 1, 100 ids)
   ├─ buffer fills to 100
   ├─ load_multi(ids)        → 1 pipeline of 100 HGETALLs
   ├─ compact ghosts          → e.g. 97 live records
   ├─ slice(25):
   │     pipelined { 25 × u.touch! }   ← 1 Redis pipeline
   │     pipelined { 25 × u.touch! }   ← 1 Redis pipeline
   │     pipelined { 25 × u.touch! }   ← 1 Redis pipeline
   │     pipelined { 22 × u.touch! }   ← 1 Redis pipeline
   └─ buffer.clear
SortedSet#each (ZSCAN page 2, 100 ids)
   └─ … repeat …
SortedSet#each exhausted
   └─ flush any remaining buffered ids the same way

Key differences

Aspect	`each`	`each_record`
Yields	raw identifier (or `[field, value]` for `HashKey`)	loaded Horreum instance
Redis ops per yield	0 extra (already paged)	amortized `HGETALL` via `load_multi` batch
Requires `record_class:` (or `class: + reference: true`)	no	yes (raises `Familia::Problem` otherwise)
Ghost handling	yields the dangling id	`compact` drops them silently
Write pipelining	not built-in	`pipeline:` groups block-body writes into `pipelined` blocks
Filters	type-specific (`since:`, `matching:`, …)	forwarded to underlying `each`

So each_record is a thin orchestration layer: it leans on the type's own each for read pagination, then layers (1) batched record hydration and (2) optional write pipelining on top.

Which collections support `each_record`?

each_record needs to know which class to hydrate. Two options supply it, and the collections Familia generates for you already set one, so each_record works on them out of the box:

ModelClass.instances — the per-class timeline. Uses class: + reference: true.
unique_index / multi_index lookups — the index hashkey/set points at the indexed class. Uses class: + reference: true.
participates_in / class_participates_in collections — point at the participant class. Use record_class:.

The two options differ in scope:

record_class: SomeClass — a loading-only hint. It enables each_record but does not change how the collection serializes/deserializes reads (members/member?/score keep the generic DataType semantics). Use this when you want each_record without any read-behavior change. This is what participates_in uses.
class: SomeClass, reference: true — a full reference type. It enables each_record and makes reads return raw-string identifiers (and member? match raw strings). Use this when you also want raw-string read semantics. This is what instances and the indexes use.

A collection you declare by hand (sorted_set :foo, set :bar, …) sets neither, so calling each_record on it raises Familia::Problem. Add record_class: (or class: + reference: true) to opt in. Note that if you pre-declare a collection that participates_in would otherwise auto-create, your hand-declared options win — add record_class: yourself if you want each_record on it.

Choosing a `pipeline` mode

each_record has two dispatch modes, controlled by pipeline:. The parameter answers a single question: may the dispatch loop wrap your block in a pipelined { }?

Value	Dispatch	Use when the block…
`nil` (default)	Each record runs in its own connection context, no pipeline wrapper	…reads, OR calls `save` / `commit_fields` / `transaction` / anything with its own internal MULTI
positive integer	Groups of `pipeline` records run inside `record_class.pipelined { ... }`	…only issues fast writers (`record.field!`) that tolerate being queued — which excludes fields backing a class-level index (those raise `Familia::IndexedFieldFastWriteError` when queued; see the indexing guide)

Note: pipeline: 0 raises ArgumentError. Use pipeline: nil to disable pipelining.

The read-only case and the serial-write case collapse into the same mode because both require immediate execution with real return values. Wrapping save in an outer pipelined would either return Redis::Future objects or raise ConflictingContextError when save's internal transaction tries to open.

The three idiomatic patterns

# 1. Read-only iteration — the default (pipeline: nil) is correct
User.instances.each_record do |user|
  puts "#{user.email} #{user.last_login}"
end

# 2. Serial writes — the default (pipeline: nil) is required for save / commit_fields / transaction
User.instances.each_record do |user|
  user.score = recompute(user)
  user.save
end

# 3. Pipelined fast writers — opt-in optimization
User.instances.each_record(pipeline: 50) do |user|
  user.last_seen_at! Familia.now   # single HSET, safe to queue in pipeline
end

Pipelining footgun

If you enable pipelining and your block reads from a related collection (e.g. user.sessions.size), that read is queued into the pipeline and returns a Redis::Future rather than a value. Omit the pipeline: parameter (or explicitly pass pipeline: nil) whenever the block needs real return values from Redis.

docs/guides/datatype-collections.md

DataType - Collection classes

Bulk writes — single round-trip mutations

Capped collections — max_length:

SortedSet: top-N by score

ListKey: per-end semantics

Atomicity

What is not capped

Enforcing the cap on existing data — enforce_max_length!

Capping a participates_in collection

Validation and the old :maxlength spelling

each — yields members (identifiers, raw strings)

each_record — yields loaded Horreum records

Concrete timeline for User.instances.each_record(batch_size: 100, pipeline: 25) { |u| u.touch! }