Check features before serving

Problem

When you need this. A job writes a feature value at 15:00. The value still describes activity before noon. Reporting its age as “just written” hides three hours of missing updates. A second error is less obvious: a row with a future cutoff has negative age and passes a simple age <= limit check.

You will test both boundaries against the store from point-in-time feature joins. You will also compare the value used for serving with historical retrieval and a recomputation from the same inputs. That comparison is training-serving parity for this fixture: the two paths agree on the feature presented to a model. There is no model invocation or prediction-quality result here.

Concepts

Freshness measures how old the represented information is at a request time. Publication recency measures when a value was written. Rewriting an old cutoff does not make its information fresh.

Training-serving parity means that two paths provide the same feature under the same data, definition and availability rules. It is a comparison contract, not a guarantee supplied by sharing a function. A freshness failure can exist while the two paths agree, and a parity failure can exist for recently written values.

Architecture

The store supplies a current projection and historical retrieval. Serving first checks cutoff and publication availability, then compares cutoff age with the policy. The parity exercise compares that result with historical selection and recomputation under a fixed snapshot; deliberately changing the arrival rule exposes skew.

Working code

Specify the age limit and the unavailable cases

The example’s feature counts the user’s events in a 24-hour window ending at cutoff. Its age at request time now is now − cutoff. The serving policy allows an age of at most 120 minutes. This is a chosen requirement for the exercise, not a limit inferred from traffic or a measured service guarantee.

The reader distinguishes four statuses. missing means there is no online row for the requested feature/entity. unavailable means a row exists but its cutoff or publication is later than the request. stale means an available row exceeds the age limit. Only fresh returns its count; all other statuses withhold value as None.

Download the complete lab and enter mlops-feature-history/. If you already unpacked the previous lesson’s ZIP, it contains this exercise too.

python3.12 -m venv .venv
source .venv/bin/activate

The code uses only the Python standard library. It runs against an in-memory SQLite database, rather than a deployed feature service.

Check availability before comparing age

read_online() looks up the feature/version and entity. It checks cutoff > now or published_at > now before testing age <= max_age. Without the first condition, a future cutoff produces negative age and appears fresh. Without the second, a value written later can appear to have existed earlier.

A fresh value at exactly the limit is accepted. One minute later it is withheld. The returned metadata still includes cutoff, publication time and age so the caller can distinguish the failure. A withheld count is never silently replaced with zero, which would assert that no event occurred.

"""Retain feature vintages; join only values published by the prediction time.

Synthetic UTC minutes and supplied publication times model a trusted writer.
This SQLite exercise is not a production event-time or publication service.
"""
import hashlib
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from vendor.availability import instant, validate, check_id
from vendor.feature_math import Feature, compute

EPOCH = datetime(2026, 1, 1, tzinfo=timezone.utc)
COUNT = Feature('user_events_1d', 1, 'user', 1, 'count')


def minute(value):
    seconds = (instant(value) - EPOCH).total_seconds()
    if seconds < 0 or seconds % 60:
        raise ValueError('fixture timestamps must be whole minutes since 2026-01-01 UTC')
    return int(seconds // 60)


def nonnegative(value):
    if type(value) is not int or not 0 <= value < 2**62:
        raise ValueError('time and age must be bounded nonnegative integer minutes')
    return value


def source_rows(events):
    validate([], events, [])
    return [{'event': index, 'user': e['entity'], 'item': 'unused',
             'event_minute': minute(e['event_at']), 'ingested_minute': minute(e['arrived_at'])}
            for index, e in enumerate(events)]


class History:
    def __init__(self, path=':memory:'):
        self.db = sqlite3.connect(path)
        self.db.row_factory = sqlite3.Row
        self.db.executescript('''
            CREATE TABLE IF NOT EXISTS definitions (
                feature TEXT PRIMARY KEY, spec TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS batches (
                feature TEXT, cutoff INTEGER, published_at INTEGER, mode TEXT NOT NULL,
                source_sha256 TEXT NOT NULL,
                PRIMARY KEY(feature, cutoff, published_at));
            CREATE TABLE IF NOT EXISTS history (
                feature TEXT, entity TEXT, cutoff INTEGER, published_at INTEGER,
                value INTEGER NOT NULL, PRIMARY KEY(feature, entity, cutoff, published_at));
            CREATE TABLE IF NOT EXISTS online (
                feature TEXT, entity TEXT, cutoff INTEGER NOT NULL, published_at INTEGER NOT NULL,
                value INTEGER NOT NULL, PRIMARY KEY(feature, entity));
        ''')

    def close(self):
        self.db.close()

    def materialize(self, events, cutoff, published_at, mode='as_known', feature=COUNT):
        cutoff = nonnegative(cutoff); published_at = nonnegative(published_at)
        if published_at < cutoff:
            raise ValueError('publication cannot precede its feature cutoff')
        if mode not in ('as_known', 'corrected'):
            raise ValueError('unknown availability mode')
        if (not isinstance(feature, Feature) or feature.entity != 'user'
                or type(feature.version) is not int or feature.version < 1
                or type(feature.window_days) is not int or feature.window_days < 1
                or feature.aggregation not in ('count', 'distinct_days')):
            raise ValueError('expected a versioned user-count or active-day feature')
        check_id(feature.name)
        # A corrected backfill still cannot use an event unavailable at publication.
        available = [e for e in source_rows(events) if e['ingested_minute'] <= published_at]
        values = compute(feature, available, cutoff, mode)
        source_hash = hashlib.sha256(json.dumps(sorted(events, key=lambda e: e['id']),
                                                sort_keys=True).encode()).hexdigest()
        key = f'{feature.name}@v{feature.version}'
        with self.db:
            old = self.db.execute('SELECT spec FROM definitions WHERE feature=?', (key,)).fetchone()
            if old and old['spec'] != feature.spec():
                raise ValueError('changed definition requires a new version')
            identity = (key, cutoff, published_at)
            batch = self.db.execute('SELECT mode,source_sha256 FROM batches WHERE feature=? '
                                    'AND cutoff=? AND published_at=?', identity).fetchone()
            if batch:
                if tuple(batch) != (mode, source_hash):
                    raise ValueError('conflicting publication identity')
                return key  # Identical retry; never rewrite a published vintage.
            latest = self.db.execute('SELECT MAX(published_at) FROM batches WHERE feature=?', (key,)).fetchone()[0]
            if latest is not None and published_at < latest:
                raise ValueError('cannot backdate publication; use the actual later availability time')
            self.db.execute('INSERT OR IGNORE INTO definitions VALUES (?,?)', (key, feature.spec()))
            self.db.execute('INSERT INTO batches VALUES (?,?,?,?,?)', (*identity, mode, source_hash))
            for entity, value in values.items():
                self.db.execute('INSERT INTO history VALUES (?,?,?,?,?)', (key, entity, cutoff, published_at, value))
                self.db.execute('''INSERT INTO online VALUES (?,?,?,?,?)
                    ON CONFLICT(feature,entity) DO UPDATE SET
                        cutoff=excluded.cutoff, published_at=excluded.published_at, value=excluded.value
                    WHERE excluded.cutoff > online.cutoff OR
                        (excluded.cutoff=online.cutoff AND excluded.published_at>online.published_at)''',
                    (key, entity, cutoff, published_at, value))
        return key

    def point_in_time(self, feature, entity, predict_at):
        nonnegative(predict_at); check_id(feature); check_id(entity)
        query = (Path(__file__).resolve().parent / 'join.sql').read_text()
        row = self.db.execute(query, (feature, entity, predict_at, predict_at)).fetchone()
        return dict(row) if row else None

    def join(self, observations, feature):
        validate(observations, [], [])
        # Preserve one row per observation, even when the feature is missing.
        return [dict(row, feature=self.point_in_time(feature, row['entity'], minute(row['predict_at'])))
                for row in observations]

    def read_online(self, feature, entity, now, max_age):
        nonnegative(now); nonnegative(max_age); check_id(feature); check_id(entity)
        row = self.db.execute('SELECT value,cutoff,published_at FROM online WHERE feature=? AND entity=?',
                              (feature, entity)).fetchone()
        if row is None:
            return {'status': 'missing', 'value': None, 'age_minutes': None}
        age = now - row['cutoff']
        if row['cutoff'] > now or row['published_at'] > now:
            status = 'unavailable'
        else:
            status = 'fresh' if age <= max_age else 'stale'
        return dict(row, status=status, value=row['value'] if status == 'fresh' else None, age_minutes=age)

The online table is a projection containing one current row per feature/entity. A later cutoff replaces an earlier one. At the same cutoff, a later publication replaces the previous vintage. Backfilling an older cutoff changes retained history but does not move this current projection backwards.

An online read at a time before the current row’s publication is therefore unavailable even if history contains an older eligible value. Use the historical join for reconstruction; this one-row projection is not an online read log. Returning an older value after an update would require a separate, explicit fallback policy.

Compare values using the same snapshot contract

The driver publishes the original noon feature at 12:30 and reads it at 13:00. It compares three values: the online row, the eligible historical row and a direct recomputation with COUNT, cutoff noon and as_known arrival filtering. Each returns one for user-01.

It then changes only the recomputation’s arrival rule to corrected. The count becomes two because the late event is now included. The mismatch is intentional. Sharing computation code reduces one cause of skew, but cannot compensate for using different source-availability rules.

"""Compare the materialized online value with the same eligible history row."""
import argparse
import json
from pathlib import Path
from history import History, COUNT, compute, source_rows, minute
from inspect_history import events, KEY, CUTOFF, FIRST, ROOT


def record():
    db=History()
    try:
        db.materialize(events(), CUTOFF, FIRST)
        now=minute('2026-01-08T13:00:00Z'); max_age=120
        online=db.read_online(KEY,'user-01',now,max_age)
        offline=db.point_in_time(KEY,'user-01',now)
        # Reference recomputation must use the *same* cutoff, arrival rule and definition.
        same_input=compute(COUNT,source_rows(events()),CUTOFF,'as_known')['user-01']
        changed_rule=compute(COUNT,source_rows(events()),CUTOFF,'corrected')['user-01']
        checks={'online':online,'historical':offline,'same_input_recomputed':same_input,
                'same_snapshot_parity':online['value']==offline['value']==same_input,
                'different_arrival_rule':changed_rule,
                'different_rule_parity':online['value']==changed_rule}
        checks['at_age_limit']=db.read_online(KEY,'user-01',CUTOFF+max_age,max_age)
        checks['past_age_limit']=db.read_online(KEY,'user-01',CUTOFF+max_age+1,max_age)
        db.materialize(events(),CUTOFF,now+60)
        checks['old_value_after_recent_write']=db.read_online(KEY,'user-01',now+120,max_age)
        checks['before_publication']=db.read_online(KEY,'user-01',now,max_age)
        # The one-row online projection cannot answer an earlier historical request.
        future_cutoff=minute('2026-01-08T16:00:00Z')
        db.materialize(events(),future_cutoff,future_cutoff+30)
        checks['future_online_row']=db.read_online(KEY,'user-01',now,max_age)
        checks['missing']=db.read_online(KEY,'unknown',now,max_age)
        return {'mode':'local SQLite; authored snapshots, no service benchmark',
                'max_age_minutes':max_age,'checks':checks}
    finally: db.close()


def main():
    p=argparse.ArgumentParser();p.add_argument('--check',action='store_true');p.add_argument('--save',type=Path)
    args=p.parse_args();result=record()
    if args.check and result!=json.loads((ROOT/'serving-result.json').read_text()):
        raise SystemExit('serving checks differ from the bundled record')
    if args.save:args.save.write_text(json.dumps(result,indent=2)+'\n')
    checks=result['checks']
    print('Online/history/same-input parity:',checks['same_snapshot_parity'])
    print('Parity after changing the arrival rule:',checks['different_rule_parity'])
    for name in ['at_age_limit','past_age_limit','old_value_after_recent_write','before_publication','future_online_row','missing']:
        row=checks[name];print(f"{name}: status={row['status']}; value={row['value']}; age={row['age_minutes']}")
    print('Age is measured from the feature cutoff; rejected reads withhold the value.')


if __name__=='__main__':main()

For a real parity check, record definition version, entity, event window, input snapshot and availability rule on both paths. If a difference appears, compare those inputs before changing the model. Equal outputs from one small fixture are a useful regression check, but do not show that every future request will match.

Run the normal, stale and future-row cases

python inspect_serving.py
python inspect_serving.py --check
python -m unittest -v test_serving.py

Online/history/same-input parity: True
Parity after changing the arrival rule: False
at_age_limit: status=fresh; value=1; age=120
past_age_limit: status=stale; value=None; age=121
old_value_after_recent_write: status=stale; value=None; age=180
before_publication: status=unavailable; value=None; age=60
future_online_row: status=unavailable; value=None; age=-180
missing: status=missing; value=None; age=None
Age is measured from the feature cutoff; rejected reads withhold the value.

At age 120, the value is returned. At age 121, it is withheld. Writing another vintage for the noon cutoff at 14:00 does not make its 15:00 read recent: age is still 180 minutes. A read at 13:00 cannot use that later publication even though its cutoff is in the past.

The final future-row probe publishes a 16:00 cutoff at 16:30, then asks the current projection about a 13:00 request. Its negative age is retained as diagnostic metadata, but status is unavailable and the count is withheld. The full result includes all selected timestamps.

Test the failure boundary separately from model behavior

test_future_cutoff_is_unavailable_and_withheld fails if a negative age becomes fresh. test_recent_write_cannot_refresh_old_feature_cutoff fails if publication time replaces cutoff in the age calculation. The parity test checks the same-snapshot agreement and the changed-rule disagreement, so a test that simply compared one code path with itself would not satisfy the contract.

These tests cover local selection behavior. They do not measure feature-pipeline availability, concurrent publication, cache invalidation or production latency. A service still needs an application decision for missing or stale features: refuse the request, use an explicitly versioned fallback, or route to a model trained for that missing-input policy. That choice belongs in its serving contract.

Design exercise: a midnight feature is rewritten at 09:00 without changing its cutoff. At 10:00, a two-hour age limit rejects it. Explain which process should be investigated first and why repeatedly rewriting the same row cannot repair the freshness violation. Then name one input difference that a parity check must detect even when both paths import the same aggregation function.

Failure cases

A future cutoff produces negative age and can pass a naive maximum-age check. Availability must refuse it before age is compared. The changed-arrival-rule case tests a different defect: both paths run, but they no longer compute the same feature contract.

Deployment

Apply the availability and freshness policy at the serving read, and retain the cutoff used for each accepted value. Define whether a missing or stale feature causes refusal, fallback or another product action before attaching a predictor. Compare historical and serving paths with identified snapshots; the fixture does not implement a production feature service.

Scaling

Measure materialization delay, publication lag and request-time feature age separately. Increasing job frequency may not repair late upstream events. A larger parity sample should cover entities, corrections and time boundaries rather than repeatedly comparing one current value.

Key takeaways

• Availability comes before freshness.
• A recent write can still contain stale information.
• Parity requires the same snapshot and rules, not merely shared code.

Best practices

• Record both cutoff and publication time in diagnostics.
• Test future and unpublished rows explicitly.
• Keep freshness failures separate from value mismatches.

References

• Feast, point-in-time joins: historical retrieval windows are relative to each observation time, and availability filtering is a separate condition.
• SQLite, conditional UPSERT: update a conflicting row only when the incoming cutoff or vintage is newer.

Sources checked October 24, 2026. The displayed outcomes are from local synthetic fixtures, with no service benchmark or model request.