Part 4 of 14

Content filtering

A reader can subscribe with a condition attached — only samples where the temperature is above a threshold, only those from one named sensor. The condition is written once, as an expression with parameters, and then has to exist in three places at once: as text the reader announces, as a compiled predicate the reader evaluates, and as the same predicate at any writer that understands it.

Keeping those three in agreement is most of the work. A condition that drifts between them does not fail loudly; it delivers a different set of samples than the one the application asked for.

Discovery

Announcing a filter

Runs when

a filtered reader is announced, or a peer's announcement is read

How often

Once per announcement. Absent from publication, history, and take entirely.

A reader tells writers about its condition so they can avoid sending what will be discarded. The announcement carries four strings — the filtered topic name, the related topic name, the filter class, and the expression — then a count of parameters and the parameters themselves.

Everything after the four strings is a count chosen by whoever sent the message, which puts this in the same category as the sequence lengths in Part 2. The count is checked twice: against the deployment's configured limit, and against whether the bytes remaining could possibly hold that many strings.

Decode-Filter-Property(b, limits)

b the borrowed parameter body · limits the configured maximum parameter count and string lengths

for each required string s in (filtered topic, related topic, class, expression)
v = Read-String(b, limits.length(s))
if v == error
return error
n = Read-32(b)
if n > limits.parameters
return error // more than this deployment will hold
if Remaining(b) < Minimum-Extent(n)
return error // n strings cannot fit in what is left
table = Reserve(n) // only now, with n proved twice
for i = 1 to n
table[i] = Read-String(b, limits.parameter-length)
if table[i] == error
return error
if Remaining(b) ≠ 0 and not all zero
return error // trailing bytes are not ignored
return property(strings, table)

Line 8 is the check that the configured limit alone does not give. A deployment allowing 64 parameters would otherwise let a twelve-byte message reserve 64 entries, because 64 passes line 6. Requiring that the remaining bytes could hold that many strings ties the reservation to what actually arrived.

Line 10 reserves after both proofs rather than between them, which is the same ordering as everywhere else: nothing is spent on input that has not finished being checked.

Cost

O(B + P) for B encoded bytes and P parameters, in time and in retained control-plane storage. None of it runs when a sample is published or taken.

Compilation

Compiling an expression

Runs when

a filtered topic is created, or its parameters change

How often

Once per condition. The compiled result is what every sample meets.

The expression is a small SQL-like language: a field, an operator, and an operand, composed with AND, OR, NOT, and parentheses. An operand is either a literal or %0, %1 and so on, referring to the parameters announced alongside it.

Compiling it rather than interpreting the text per sample is the obvious move. The part that is less obvious is what compilation must do with grammar it does not support.

Why an unsupported expression must not compile

The tempting behaviour for an unrecognised construct is to ignore it and evaluate the rest. That produces a filter that is weaker than the one written — it admits samples the application excluded, and it does so silently. So an unknown field, an unknown operator, a missing parameter, a malformed range, or a trailing token produces no predicate at all.

Failing to compile is safe here precisely because of the division of labour in the last section of this part: a writer that cannot compile a condition sends everything, and the reader — which is authoritative — applies it.

Compile(text, params, fields)

text the expression · params the announced parameter values · fields the field names the generated type declares

if not Quotes-Balanced(text) or not Parens-Balanced(text)
return invalid
return Parse(text, params, fields)

Parse(t, params, fields)

t one expression or sub-expression · params, fields as above

if t is empty
return invalid // an empty branch is not "always true"
if t has a top-level or
(l, r) = Split-At-Top-Level(t, or)
return either(Parse(l, …), Parse(r, …))
if t has a top-level and // not the one inside a range
(l, r) = Split-At-Top-Level(t, and)
return both(Parse(l, …), Parse(r, …))
if t begins with not
return complement(Parse(rest of t, …))
return Parse-Leaf(t, params, fields)

The order on lines 3, 6 and 9 is the precedence: splitting at OR first makes it bind loosest, which is what SQL means. Splitting at AND first would silently reparenthesise every expression that mixes them.

The comment on line 6 is a real ambiguity in the grammar. A range is written field BETWEEN lower AND upper, so the scanner meets an AND that is part of the range rather than a Boolean conjunction. Treating it as a conjunction splits the range in half and produces two nonsense leaves.

Worked example

temp > %0 AND zone BETWEEN 3 AND 7, with parameter 0 set to 20:

Step	Result
Top-level `OR`?	none
Top-level `AND`?	the first one only — the second belongs to `BETWEEN`
Left leaf	`temp > 20`, with `%0` resolved
Right leaf	`3 ≤ zone ≤ 7`

A scanner that took the second AND would produce zone BETWEEN 3 and 7 — the first malformed, the second not a predicate at all. Both are rejections, so the failure is at least loud; the danger is a scanner that accepts one of the halves.

Cost

Compilation is control-plane work proportional to the expression length. String literals are borrowed from the expression rather than copied, except one that contains an escaped quote, which is decoded once. Evaluating the result later touches only the nodes short-circuit evaluation reaches.

Composition

Composing predicates in code

Runs when

an application builds a filter from typed pieces rather than an expression

How often

Construction is once; the composed tree is evaluated per sample.

The alternative to writing an expression is building the condition from typed pieces — a comparison on one field, membership in a set, combined with and, or, and negation. This gives the compiler a chance to check field names and types, which a string cannot.

Composing raises two costs that a single hand-written predicate would not have, and both are dealt with at construction rather than per sample.

Deserializing once for the whole tree

The naive composition gives each child the serialized bytes and lets it decode what it needs. A tree of six comparisons then decodes the sample six times. Instead the whole tree is wrapped once: the sample is decoded a single time and every child evaluates against that one value.

Sorting a set once, not scanning it per sample

Membership in a set of N values, checked by walking the set, is N comparisons on every sample forever. The values are known when the predicate is built, so the array is sorted then, and each later query is a binary search.

Evaluate(node, v)

node one node of the composed tree · v the sample, decoded once by the caller

if node.kind == compare
return Compare(node.field(v), node.op, node.operand)
if node.kind == in-set
return Binary-Search(node.sorted, node.field(v)) ≠ nil
if node.kind == both
return Evaluate(node.left, v) and Evaluate(node.right, v) // short-circuits
if node.kind == either
return Evaluate(node.left, v) or Evaluate(node.right, v)
return not Evaluate(node.child, v)

Cost

Building a set costs O(N log N) once and O(log N) per query afterwards, in an array the caller supplied so nothing is allocated for it. A sample is decoded once, whatever the shape of the tree, and only the nodes short-circuiting reaches are evaluated.

Admission

Applying a reader's condition

Runs when

a sample reaches a reader that has a condition attached

How often

Once per sample per filtered reader, before it is stored. Changing the condition is a control-plane action and is rare.

A reader can subscribe to a topic with a condition attached: only samples where the temperature is above a threshold, only those from a named sensor. The condition is written once as an expression with parameters, and both sides need it — the reader to apply it, and the writer to learn it through discovery so it can avoid sending what will be discarded.

Where the condition is applied matters more than it looks. If a sample is stored first and filtered when the application reads it, then a rejected sample has already taken a slot in a bounded history, and it may have pushed out a sample the reader wanted. So the condition runs at admission, before the sample is stored.

Every route into the history

The same test has to sit on every route into the history. A sample can come from the network, from a writer in the same process, or through shared memory, and a filter placed on only one of those is not a filter — it is a filter that works until the deployment changes.

Figure 4.1 Every route into the history passes the same test, before the sample occupies a slot.

Admit-Sample(reader, sample)

reader the destination, holding its condition and its history · sample the arriving sample, from any of the three routes

if reader.filter ≠ nil
g = Read-Lock(reader.filter) // one complete generation
keep = Evaluate(g.predicate, sample)
Unlock(g)
if not keep
return rejected // history is untouched
return Store(reader.history, sample)

Line 1 is the whole cost for a reader with no condition: one check for absence, and no lock is taken.

Changing the condition while samples arrive

The condition can be changed while samples are arriving, which is where the generation number comes in. The expression, its parameters, and the compiled form of both must never be observed half-replaced — a sample evaluated against the new expression with the old parameters is neither the old filter nor the new one. So the candidate is fully compiled before any lock is taken, and all three are swapped together inside it.

Update-Filter(f, expression, params)

f the shared filter state — parameters, compiled form, and generation · expression, params the new condition

if not Valid(expression) or |params| > f.limit
return error // nothing retained
p = Compile(expression, params) // the slow part, outside the lock
if p == error
return error
Write-Lock(f)
f.generation = f.generation + 1
f.params = params
f.predicate = p
Unlock(f)
return f.generation

An announcement that arrives late

The generation number is not only for local consistency. The reader announces its condition to writers, and announcements can be reordered or repeated. A writer accepts an announcement only if its generation is higher than the one it already holds, so a delayed old announcement cannot roll a writer back to a stale condition. A writer that receives a condition it cannot evaluate does not guess: it sends everything, which is always correct because the reader applies the condition again anyway.

Cost

A filtered sample costs one lock acquisition plus O(S) to deserialize and evaluate it, for S bytes. An unfiltered reader costs one check for absence. Changing a condition costs O(E + P + B) for the expression and parameter bytes, and none of it happens inside the lock except the swap.

Update

Replacing a filter

Runs when

an application changes the condition on a reader that already has samples queued

How often

Once per change, scanning what is currently queued.

Changing the condition raises a question the first installation does not: what about the samples already waiting? Applying the new condition only to later arrivals leaves the application holding samples that its current condition excludes.

So the queued samples are re-tested against the new condition and the rejections removed — and only then is the new condition installed, so no sample is ever tested against a condition that is partly installed.

Replace-Filter(reader, f)

reader with its retained history and, when present, its shared-memory history · f the new condition

for each sample s in reader.history // queue order
if not Evaluate(f, s)
Remove(reader.history, s)
for each queued descriptor d in reader.shm-history
bytes = Borrow(d) // borrow, not consume — the reference stays
if not Evaluate(f, d.info, bytes)
Remove(reader.shm-history, d)
Release(d) // exactly one reference, the one this reader held
reader.filter = f // installed only after both histories are pruned

Line 5 borrows the shared-memory payload rather than taking it. This is the distinction from Part 5: taking would consume the reader's reference, and a sample that then passes the new condition would be left queued with no reference to the bytes it names.

Line 6 evaluates against the metadata stored beside the descriptor rather than looking the sample's identity up in the endpoint graph. The writer may have gone since the sample was queued, and a sample that was admissible when it arrived should not become unevaluatable because its writer left.

Cost

One scan of the Q queued samples, borrowing each payload in place with no copy. An unfiltered reader never runs it.

Egress

Filtering at the writer

Runs when

a writer publishes to a reader whose condition it was able to compile

How often

Once per sample per such reader.

A writer that understands a reader's condition can avoid sending what will be discarded, saving the bandwidth and the reader's work. This is the only part of filtering that is an optimisation rather than a requirement, and it is the part with a trap in it.

A reliable reader tracks sequence numbers. Simply not sending sample 15 leaves a hole, and the reader does exactly what Part 3 says it does: notices the gap and asks for it, indefinitely. Filtering at the writer without telling the reader converts a bandwidth saving into a permanently stalled association.

So a filtered-out sample is not silence. The writer declares that sequence irrelevant, which is the same mechanism Part 3 describes for a sample dropped from history, and follows it with its usual announcement so the reader learns where things now stand.

Figure 4.2 Not sending is not the same as saying nothing.

Publish-To-Reader(w, r, sample)

w the writer · r one matched reader, with the compiled form of its condition when the writer could compile it · sample the serialized sample

if r.compiled == nil
return Send-Data(w, r, sample) // cannot evaluate it, so send
if Evaluate(r.compiled, sample)
return Send-Data(w, r, sample)
Send-Irrelevant(w, r, sample.seq) // close the hole
return Send-Announcement(w, r)

Line 2 is the fallback, and it is always correct rather than merely acceptable. The reader applies the condition again at admission regardless, so a writer that sends too much costs bandwidth and nothing else. A writer that sends too little would cost correctness, which is why the failure direction is fixed this way and why the compiler in the second section refuses rather than approximates.

Cost

One predicate evaluation per sample per reader whose condition the writer compiled. A rejection costs two small control records instead of the sample, which is the saving whenever the sample is larger than about a hundred bytes.

Next: history, durability and instances

Where a sample is stored, and what a reader that starts an hour late receives.

Part 5 →

These descriptions follow the implementations in Tin DDS and Retina. Evaluation access and the full algorithm notes are available from engineering@tinverse.com.