Constraints and Monitoring

Constraints let you define invariants over your event stream: rules that must hold or situations that must never occur. Combined with the ConstraintMonitor, they enable real-time verification of system behavior.

must_match

The must_match constraint asserts that a given pattern must eventually be satisfied. If the computation completes without the pattern matching, a ConstraintViolation is raised.

must_match.py python
from pyrapide import must_match, Pattern

# Every request must eventually produce a response
must_match(
    Pattern.match("api.request") >> Pattern.match("api.response"),
    name="request-response-completeness"
)

never

The never constraint asserts that a pattern must not match at any point during execution. If it does, a violation is raised immediately.

never.py python
from pyrapide import never, Pattern

# Data must never be written after a deletion event
never(
    Pattern.match("record.deleted") >> Pattern.match("record.written"),
    name="no-write-after-delete"
)

# Two critical sections must never overlap (must be independent)
never(
    Pattern.match("lock.acquired") | Pattern.match("lock.acquired"),
    name="no-concurrent-locks"
)
⚠ Warning
never constraints are checked eagerly. The monitor evaluates them on every new event, so violations are caught as soon as they occur, not at the end of the computation.

The @constraint Decorator

For more complex invariants that require custom logic, use the @constraint decorator. It wraps a function that receives matched events and returns a boolean indicating whether the constraint holds.

constraint_decorator.py python
1from pyrapide import constraint, Pattern
2
3@constraint(
4    pattern=Pattern.match("transaction.complete"),
5    name="transaction-balance"
6)
7def check_balance(match):
8    """Verify that debits and credits balance after every transaction."""
9    event = list(match.events)[0]
10    debits = event.payload.get("total_debits", 0)
11    credits = event.payload.get("total_credits", 0)
12    return abs(debits - credits) < 0.01
13
14@constraint(
15    pattern=Pattern.match("api.response"),
16    name="response-time-sla"
17)
18def check_sla(match):
19    """Responses must arrive within the SLA window."""
20    event = list(match.events)[0]
21    latency = event.payload.get("latency_ms", 0)
22    return latency < 500

ConstraintMonitor

The ConstraintMonitor watches a live Computation and evaluates all registered constraints in real time. It provides hooks for violation handling, logging, and alerting.

monitor.py python
1from pyrapide import ConstraintMonitor, Computation
2
3comp = Computation()
4monitor = ConstraintMonitor(comp)
5
6# Register constraints
7monitor.add(must_match(
8    Pattern.match("healthcheck") ,
9    name="healthcheck-present"
10))
11
12monitor.add(never(
13    Pattern.match("data.corrupted"),
14    name="no-corruption"
15))
16
17# Set a violation handler
18@monitor.on_violation
19def handle_violation(violation):
20    print(f"VIOLATION: {violation.constraint_name}")
21    print(f"  Events involved: {violation.events}")
22    print(f"  Timestamp: {violation.detected_at}")
23
24# Start monitoring (the monitor evaluates on each recorded event)
25monitor.start()

Constraint Composition

Constraints can be grouped and managed as a set. This is useful for applying different constraint profiles to different environments (e.g., strict constraints in testing, relaxed in production).

constraint_set.py python
from pyrapide import ConstraintSet

strict = ConstraintSet("production-invariants")
strict.add(must_match(
    Pattern.match("api.request") >> Pattern.match("api.response"),
    name="completeness"
))
strict.add(never(
    Pattern.match("unhandled.exception"),
    name="no-unhandled-exceptions"
))
strict.add(check_balance)
strict.add(check_sla)

# Apply the entire set to a monitor
monitor.add_set(strict)
💡 Tip
Use ConstraintSet to organize invariants by domain (e.g., "security", "performance", "data-integrity"). This makes it easy to enable or disable groups of checks depending on the deployment environment.

Violation Reports

When a constraint is violated, the monitor produces a ConstraintViolation object containing:

  • constraint_name: the name of the violated constraint.
  • events: the set of events involved in the violation.
  • detected_at: the logical timestamp when the violation was detected.
  • pattern: the pattern that triggered (or failed to trigger) the constraint.
  • message: a human-readable description of the violation.
violation_handler.py python
@monitor.on_violation
def alert(violation):
    if violation.constraint_name == "response-time-sla":
        send_pager_alert(
            message=f"SLA breach: {violation.message}",
            events=violation.events,
        )

Constraints rely on Clocks for timed patterns and temporal ordering. Learn about clock types next.