GitHub - damienos61/rift: Causal execution engine for Go — eliminate Heisenbugs by turning non-determinism into a controllable resource

// Without Rift — classic race condition
var counter int64

go func() { counter++ }()
go func() { counter++ }()

// What is counter now? 1 or 2?
// It depends on the goroutine scheduler — which you cannot control.
// Under the race detector: likely 2. In production: sometimes 1.
// This bug disappears when you add logging. It's a Heisenbug.

// With Rift — same function, deterministic result
eng, _ := rift.NewEngine(rift.DefaultConfig())

result, err := eng.Run(func() (any, error) {
    counter := atomic.AddInt64(&sharedCounter, 1)
    return counter, nil
})
// result.Value is always the causally-correct answer.
// No rewrite needed. No locks added to your code.

Your function
      │
      ▼
┌─────────────┐
│   SPLIT     │  Clone fn into N goroutines (RiftFactor, default: 3)
└─────────────┘
      │
      ├──▶  Rift A  ──▶ [runs fn] ──▶ result A + ECC-2 clock A ──▶ ╮
      ├──▶  Rift B  ──▶ [runs fn] ──▶ result B + ECC-2 clock B ──▶ ──▶ ┌──────────┐
      └──▶  Rift C  ──▶ [runs fn] ──▶ result C + ECC-2 clock C ──▶ ╯   │  FUSE    │──▶ Deterministic Result
           (all parallel, no global locks, panics caught)               └──────────┘

go get github.com/damienos61/rift

package main

import (
    "fmt"
    "github.com/damienos61/rift"
)

func main() {
    // DefaultConfig gives you sensible production defaults:
    // RiftFactor=3, timeout=5s, fusion strategy="causal"
    eng, err := rift.NewEngine(rift.DefaultConfig())
    if err != nil {
        panic(err)
    }
    // eng is safe to use from multiple goroutines simultaneously.

    result, err := eng.Run(func() (any, error) {
        // Put any function here — DB query, HTTP call, computation, etc.
        // Rift will run it 3 times in parallel and pick the best result.
        return "hello from rift", nil
    })

    if err != nil {
        fmt.Println("all variants failed:", err)
        return
    }

    fmt.Println(result.Value)        // "hello from rift"
    fmt.Println(result.CausalScore)  // e.g. 3.4104 — ECC-2 priority score
    fmt.Println(result.EntropyDelta) // e.g. 0.12   — path diversity delta
    fmt.Println(result.FusedFrom)    // e.g. 7      — which rift ID won
    fmt.Println(result.Retries)      // 0           — retries used
}

git clone https://github.com/damienos61/rift
cd rift
go run ./cmd/rift              # basic demo — 8 operations, shows scores and entropy
go run ./cmd/rift -heisen      # Heisenbug simulation with 200 goroutines
go run ./cmd/rift -bench       # throughput benchmark

╔══════════════════════════════════════════════════════════╗
║       R I F T  v1.3.0 — Causal Execution Engine         ║
╚══════════════════════════════════════════════════════════╝
  RiftFactor : 3  |  Strategy : causal

  op=1 | value=result-1     | score=3.3113 | entropy=0.1204 | retries=0
  op=2 | value=result-2     | score=3.4104 | entropy=0.0891 | retries=0
  ...

─── Metrics ─────────────────────────────────────────────
  ops:     total=8  success=8  failed=0  shed=0
  rifts:   spawned=24  pruned=16
  engine:  latency=0.02ms  entropy=0.1031  error_rate=0.0%
  circuit: closed  active_factor=3  strategy=causal

type CausalClock struct {
    RiftID     RiftID   // unique ID of this rift variant
    Lamport    uint64   // logical event counter (like a Lamport timestamp)
    Weight     float64  // the main priority score — computed by Finalize()
    WallNanos  int64    // wall clock time when this rift finished (nanoseconds)
    Entropy    float64  // execution path diversity score in (0, 1)
    Depth      uint32   // how many causal events this rift observed
    Generation uint32   // monotonic counter — prevents clock confusion under churn
}

ratio = latencyBudget / actualDuration

if ratio >> 1 (much faster than budget) → bonus ~2.0
if ratio == 1 (exactly on budget)       → neutral ~1.0
if ratio << 1 (much slower)             → penalty ~0.5

Weight *= 0.5 + 1.5 / (1 + exp(-2.2 × (ratio - 1)))

if healthy:    Weight *= 1.55   (+55% boost)
if failed:     Weight *= 0.08   (-92% penalty)

if Lamport > 1:
    Weight *= 1.0 + log1p(Lamport) × 0.05

entropy = normEntropy(pathHash XOR entropyPool)   -- result in (0, 1)
Weight *= 1.0 + 0.15 × entropy

if Depth > 0:
    Weight *= 1.0 + log1p(Depth) × 0.02

Score = Weight × (1 + 0.15 × Entropy)
      + log1p(Lamport) × 0.01
      + log1p(Depth)   × 0.02
      + 1 / (1 + WallNanos / 1e9) × 0.001

EntropyDelta = winner.Entropy - mean(loser1.Entropy, loser2.Entropy, ...)

CLOSED ──(error rate ≥ threshold)──▶ OPEN ──(after cool-down)──▶ HALF-OPEN
  ▲                                                                    │
  │                                                         (probe OK) │
  └────────────────────────────────────────────────────────────────────┘
                                                         (probe fails) │
                                                                ▼
                                                              OPEN (again)

cfg.CircuitBreaker = rift.CircuitBreakerConfig{
    Enabled:   true,
    Threshold: 0.5,              // open when 50% of ops in the window fail
    Window:    100,              // rolling window: last 100 operations
    CoolDown:  10*time.Second,   // wait 10s before probing
}

go run ./cmd/rift -circuit

═══ Circuit Breaker Demo ═══
  Injecting failures to trip the breaker...
  op=01 err=true  circuit=closed
  op=02 err=true  circuit=closed
  ...
  op=10 err=true  circuit=open
  op=11 err=true  circuit=open    ← engine stops trying
  ...
  Post-saturation: err=true (expect circuit-open)
  Waiting 2s for cool-down...
  Probe OK: recovered  circuit=closed  ← back to normal

cfg.Retry = rift.RetryPolicy{
    MaxAttempts: 3,                       // try up to 3 times per rift
    Backoff:     1*time.Millisecond,      // start with 1ms delay
    MaxBackoff:  50*time.Millisecond,     // cap the delay at 50ms
}
// Actual delays: attempt 2 waits 1ms, attempt 3 waits 2ms, capped at 50ms

Attempt 1: run immediately
Attempt 2: wait Backoff (1ms), then run
Attempt 3: wait min(Backoff×2, MaxBackoff) = min(2ms, 50ms) = 2ms, then run

cfg.Adaptive = rift.AdaptiveConfig{
    Enabled:       true,
    MinFactor:     2,    // never go below 2 rifts (minimum for fusion to work)
    MaxFactor:     8,    // never go above 8 rifts
    ErrorRateUp:   0.3,  // if error rate > 30%, increase RiftFactor by 1
    ErrorRateDown: 0.05, // if error rate < 5%,  decrease RiftFactor by 1
}

go run ./cmd/rift -adaptive

  Phase: healthy      (fail_rate=0%)
  active_factor=2  error_rate=0.0%  circuit=closed

  Phase: degraded     (fail_rate=60%)
  active_factor=6  error_rate=42.0%  circuit=closed   ← scaled up under stress

  Phase: recovering   (fail_rate=10%)
  active_factor=5  error_rate=30.0%  circuit=closed

  Phase: healthy      (fail_rate=0%)
  active_factor=2  error_rate=5.0%   circuit=closed   ← scaled back down

cfg.Shed = rift.ShedPolicy{
    Enabled:     true,
    MaxQueueLen: 1000,    // allow up to 1000 in-flight operations
    Strategy:    "newest", // currently: reject new arrivals when full
}

result, err := eng.Run(myFn)
if errors.Is(err, rift.ErrShed) {
    // Operation was dropped — queue your retry logic here,
    // or return 503 to the caller
    return
}

═══ Load-Shedding Demo (v1.3.0) ═══
  Sending 500 ops into an engine with MaxQueueLen=50.
  processed=50  shed=450  engine_shed_count=450

cfg.Warmup = rift.WarmupConfig{
    Enabled: true,
    Ops:     10,               // run 10 no-op operations at startup
    Timeout: 2*time.Second,    // if warmup doesn't finish in 2s, fail with ErrWarmupTimeout
}

eng, err := rift.NewEngine(cfg)
if errors.Is(err, rift.ErrWarmupTimeout) {
    // Warmup timed out — the system may be under severe load at startup
}

h := eng.Health()

h.Live   // bool — true if the engine is running (always true if you have an Engine)
h.Ready  // bool — true if: (1) circuit is closed AND (2) error rate ≤ 50%
h.Reason // string — human-readable reason when Ready=false, e.g. "circuit breaker open"

http.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
    h := eng.Health()
    if !h.Live {
        http.Error(w, "engine not live", 503)
        return
    }
    w.WriteHeader(200)
    fmt.Fprintln(w, "live")
})

http.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
    h := eng.Health()
    if !h.Ready {
        http.Error(w, h.Reason, 503)
        return
    }
    w.WriteHeader(200)
    fmt.Fprintln(w, "ready")
})

go run ./cmd/rift -health

═══ Health Probe Demo (v1.3.0) ═══
  Healthy engine  — live=true  ready=true
  Stressed engine — live=true  ready=false  reason="circuit breaker open"

type TelemetryHook interface {
    // Called once per Run() call, right after the split.
    OnSplit(opID OperationID, factor int)

    // Called once per rift, immediately after it finishes (success or error).
    OnConverge(riftID RiftID, duration time.Duration, healthy bool)

    // Called once per Run() call, after fusion selects a winner.
    OnFuse(opID OperationID, winner RiftID, score float64, pruned int)

    // Called when the fused result is an error.
    OnError(opID OperationID, err error)

    // Called when an operation is dropped by the load shedder (v1.3.0).
    OnShed(opID OperationID)
}

type myTracer struct{}

func (t *myTracer) OnSplit(opID rift.OperationID, factor int) {
    mySpan.Start(fmt.Sprintf("rift.split op=%d factor=%d", opID, factor))
}
func (t *myTracer) OnConverge(riftID rift.RiftID, d time.Duration, healthy bool) {
    myMetrics.Record("rift.converge.latency_ms", d.Milliseconds())
}
func (t *myTracer) OnFuse(opID rift.OperationID, winner rift.RiftID, score float64, pruned int) {
    myMetrics.Record("rift.causal_score", score)
}
func (t *myTracer) OnError(opID rift.OperationID, err error) {
    myAlerts.Trigger("rift.operation_failed", err)
}
func (t *myTracer) OnShed(opID rift.OperationID) {
    myMetrics.Increment("rift.shed_total")
}

cfg := rift.DefaultConfig()
cfg.Telemetry = &myTracer{}
eng, _ := rift.NewEngine(cfg)

cfg := rift.DefaultConfig()  // always start from defaults

// ── Core ──────────────────────────────────────────────────────────────────────
cfg.RiftFactor     = 3                // Number of parallel variants [2, 32]
cfg.WorkerPoolSize = 0                // Goroutine pool size. 0 = 2×GOMAXPROCS
cfg.DefaultTimeout = 5*time.Second    // Per-rift timeout. 0 = no timeout.

// FusionStrategy selects how the winner is chosen:
//   "causal"   (default) — ECC-2 total order (Weight, Entropy, Lamport, Wall, ID)
//   "entropy"  — Entropy is primary; useful for chaos testing
//   "lamport"  — Pure Lamport counter ordering (classic vector clock)
//   "fastest"  — Lowest wall time wins (non-deterministic; benchmarks only)
cfg.FusionStrategy = "causal"

// ── Circuit breaker ──────────────────────────────────────────────────────────
cfg.CircuitBreaker = rift.CircuitBreakerConfig{
    Enabled:   false,             // disabled by default
    Threshold: 0.5,               // open at 50% error rate
    Window:    100,               // rolling window size (number of operations)
    CoolDown:  10*time.Second,    // half-open probe delay after opening
}

// ── Retry ────────────────────────────────────────────────────────────────────
cfg.Retry = rift.RetryPolicy{
    MaxAttempts: 1,               // 1 = no retry (default). Set ≥ 2 for retries.
    Backoff:     1*time.Millisecond,
    MaxBackoff:  50*time.Millisecond,
}

// ── Adaptive RiftFactor ──────────────────────────────────────────────────────
cfg.Adaptive = rift.AdaptiveConfig{
    Enabled:       false,         // disabled by default
    MinFactor:     2,
    MaxFactor:     6,
    ErrorRateUp:   0.3,           // increase factor when error rate > 30%
    ErrorRateDown: 0.05,          // decrease factor when error rate < 5%
}

// ── Load shedding ────────────────────────────────────────────────────────────
cfg.Shed = rift.ShedPolicy{
    Enabled:     false,           // disabled by default
    MaxQueueLen: 1000,            // max concurrent in-flight operations
    Strategy:    "newest",        // "newest": reject new arrivals when full
}

// ── Warmup ───────────────────────────────────────────────────────────────────
cfg.Warmup = rift.WarmupConfig{
    Enabled: false,               // disabled by default
    Ops:     10,                  // number of no-op warmup operations
    Timeout: 2*time.Second,       // max time allowed for warmup
}

// ── Telemetry ────────────────────────────────────────────────────────────────
cfg.Telemetry = nil               // nil = no telemetry (default). Set to your hook.

eng, err := rift.NewEngine(cfg)

# Basic demo — 8 operations showing score, entropy, and metrics
go run ./cmd/rift

# With a custom RiftFactor and fusion strategy
go run ./cmd/rift -factor 5 -strategy entropy

# Heisenbug simulation — 200 concurrent goroutines racing on a counter
go run ./cmd/rift -heisen

# Throughput benchmark — 20,000 operations, 8 workers
go run ./cmd/rift -bench

# Circuit breaker demo — inject failures, watch circuit open and recover
go run ./cmd/rift -circuit

# Adaptive RiftFactor demo — 4 phases of varying error rates
go run ./cmd/rift -adaptive

# Load-shedding demo — 500 ops into a MaxQueueLen=50 engine
go run ./cmd/rift -shed

# Health probe demo — healthy vs stressed engine comparison
go run ./cmd/rift -health

go run ./examples/trading

# Run all 22 tests
go test ./...

# Run with the race detector — must produce zero race warnings
go test -race ./...

# Run a specific test with verbose output
go test -run TestHeisenBugElimination -v
go test -run TestCircuitBreakerClosesAfterCoolDown -v
go test -run TestLoadSheddingRejectsWhenSaturated -v

# Run all benchmarks with memory stats
go test -bench=. -benchmem

# Run benchmarks for 5 seconds each (more stable numbers)
go test -bench=. -benchtime=5s -benchmem

rift/
│
├── rift.go              # Public API — the only file you need to import
│                        # Re-exports all types; wraps the internal engine.
│
├── rift_test.go         # 22 tests + 6 benchmarks (all in package rift_test)
│
├── go.mod               # Module: github.com/damienos61/rift, go 1.22
├── README.md            # This file
├── CHANGELOG.md         # Version history with detailed change notes
├── LICENSE              # MIT
├── .gitignore
│
├── cmd/
│   └── rift/
│       └── main.go      # Interactive CLI — all demos and flags
│
├── examples/
│   └── trading/
│       └── main.go      # HFT order processor — all features at once
│
└── internal/            # Internal packages — not part of the public API
    │
    ├── rift/
    │   ├── types.go     # All shared types: Config, Result, CausalClock,
    │   │                # State, Operation, all interfaces, sentinel errors
    │   └── rift.go      # Rift struct + state machine + FNV-1a path hash
    │
    ├── clock/
    │   └── clock.go     # ECC-2 implementation: Tick, Finalize, Compare, Score
    │                    # The core algorithm — see the deep dive section above
    │
    ├── splitter/
    │   └── splitter.go  # Creates N Rift objects from one Operation
    │                    # Closure-safe capture, MaxRiftFactor=32 guard
    │
    ├── executor/
    │   └── executor.go  # Runs rifts concurrently, manages the semaphore pool,
    │                    # handles retry backoff, catches panics
    │
    ├── fusion/
    │   └── fusion.go    # 3-pass selection: filter → rank → select+prune
    │                    # All 4 strategies: causal, entropy, lamport, fastest
    │
    └── engine/
        └── engine.go    # Top-level orchestrator: wires everything together
                         # Circuit breaker, adaptive factor, shedder, metrics,
                         # warmup, health probe