## The problem that got me curious

I hit a weird failure in a CI pipeline that “looked secure” on paper: the pipeline was verifying artifacts, but it was implicitly trusting the certificate that signed the artifact identity.

That’s the core issue in software supply chain security: *identity for artifacts is established via signatures and certificates*, and if you don’t verify the certificate chain the right way, you can end up trusting the wrong signer—even when signature verification “passes”.

I wanted to understand how to *preemptively* validate provenance produced with **Sigstore** (a Sigstore stack commonly used to sign artifacts) by verifying the **Fulcio** certificate that issues signing certificates. Fulcio is the service that issues short-lived signing certificates; those certificates chain up to a known trust root. If that chain isn’t validated correctly, provenance checks become fragile.

This post documents exactly what I built: a small CI-friendly verifier that checks:

- the artifact was signed (using Sigstore’s Rekor transparency log),
- the signing certificate’s chain is valid and anchored to an allowlisted Fulcio root,
- and the certificate’s subject aligns with expected workflow identity claims.

## What I built (high level)

I implemented a command-line verifier in Node.js that:

1. Fetches the certificate information from the Sigstore signature payload.
2. Validates the certificate’s X.509 chain back to a known Fulcio trust anchor.
3. Checks an expected identity constraint (for example, “URI contains this GitHub repository and workflow ref”).
4. Fails the build if any of the checks don’t line up.

This is narrow and practical: it targets the most common blind spot I saw—**skipping certificate trust validation** or validating it too loosely.

---

## Test setup I used (minimal but real)

I used a simple Sigstore signing flow with a **short-lived certificate from Fulcio** and stored signatures/claims in **Rekor**. Then I verified them in a separate CI step.

To keep this post focused, the code below assumes you already have:

- an artifact URL or local path (for digest),
- a Sigstore bundle (signature + certificate-related info),
- and the Fulcio root certificate you consider trustworthy.

### Why an allowlisted Fulcio root matters

Fulcio is like a certificate “issuer”. The verifier must trust Fulcio’s root (or intermediate) certificates. If you trust whatever certificate appears in the bundle, you’ve moved the trust boundary to the attacker.

---

## The verifier code (Node.js)

### Install dependencies

```bash
npm init -y
npm i node-forge jose
```

- **node-forge**: parses and validates X.509 certificates.
- **jose**: handles COSE/JWS-like payload parsing when needed (some Sigstore formats embed JWT/claims).

### `verify-sigstore-fulcio-chain.js`

```javascript
#!/usr/bin/env node
/**
 * Verify Sigstore signature provenance by validating:
 * 1) X.509 certificate chain rooted at an allowlisted Fulcio trust anchor
 * 2) an identity claim in the certificate (URI-based) matches an expected prefix
 *
 * Usage:
 *   node verify-sigstore-fulcio-chain.js \
 *     --bundle sigstore-bundle.json \
 *     --fulcio-root fulcio-root.pem \
 *     --identity-uri-prefix "https://token.actions.githubusercontent.com:443/*"
 */

const fs = require("fs");
const path = require("path");
const forge = require("node-forge");

function parseArgs() {
  const args = process.argv.slice(2);
  const out = {};
  for (let i = 0; i < args.length; i++) {
    const k = args[i];
    if (!k.startsWith("--")) continue;
    const key = k.slice(2);
    out[key] = args[i + 1];
    i++;
  }
  return out;
}

function pemToCert(pem) {
  return forge.pki.certificateFromPem(pem);
}

function normalizePem(input) {
  // Allow passing a file that contains PEM with or without trailing whitespace.
  return input.replace(/\r\n/g, "\n").trim() + "\n";
}

function certificateChainValidate({
  leafCertPem,
  intermediateCertsPem,
  fulcioRootPem,
}) {
  const leaf = pemToCert(normalizePem(leafCertPem));
  const root = pemToCert(normalizePem(fulcioRootPem));

  const intermediates = (intermediateCertsPem || [])
    .filter(Boolean)
    .map((pem) => pemToCert(normalizePem(pem)));

  // Build a simple trust chain:
  // leaf -> (one of intermediates) -> root
  // This example keeps things explicit and strict.
  //
  // In real Sigstore bundles, you may need to parse and extract intermediates
  // depending on the signing/cert material included.
  const candidates = [root, ...intermediates];

  // node-forge doesn't have a one-liner "validate this chain" exactly like OpenSSL,
  // so we verify signatures step-by-step.
  //
  // 1) Verify leaf signature with each candidate's public key.
  // 2) If an intermediate verifies leaf, verify intermediate signature with root.
  let leafVerifiedBy = null;
  for (const candidate of candidates) {
    const verified = leaf.verify(candidate.publicKey);
    if (verified) {
      leafVerifiedBy = candidate;
      break;
    }
  }

  if (!leafVerifiedBy) {
    throw new Error("Leaf certificate signature could not be verified against supplied trust candidates.");
  }

  const isDirectToRoot = leafVerifiedBy === root;

  if (isDirectToRoot) {
    // If the leaf is directly signed by root, chain is OK.
    // We still want to check validity time for sanity.
  } else {
    // Validate that the intermediate is signed by the root.
    const intermediateVerified = leafVerifiedBy.verify(root.publicKey);
    if (!intermediateVerified) {
      throw new Error("Intermediate certificate signature could not be verified against Fulcio root.");
    }
  }

  // Check time validity: notBefore <= now <= notAfter
  const now = new Date();
  const notBefore = leaf.validity.notBefore;
  const notAfter = leaf.validity.notAfter;

  if (now < notBefore || now > notAfter) {
    throw new Error(`Leaf certificate is not valid at current time. now=${now.toISOString()} notBefore=${notBefore} notAfter=${notAfter}`);
  }

  return { chain: "leaf->...->fulcio-root", isDirectToRoot };
}

function extractCertsFromBundle(bundle) {
  /**
   * This is the part that depends on bundle format.
   *
   * Sigstore commonly uses certificate material embedded in a structure.
   * For this example, I assumed a JSON bundle with fields:
   * {
   *   "cert": { "pem": "-----BEGIN CERTIFICATE-----..." },
   *   "intermediates": [ { "pem": "..." }, ...]
   * }
   *
   * If your bundle differs, you’d adapt this extraction layer.
   */
  if (!bundle || !bundle.cert || !bundle.cert.pem) {
    throw new Error("Bundle is missing bundle.cert.pem");
  }
  const leafCertPem = bundle.cert.pem;

  const intermediateCertsPem = Array.isArray(bundle.intermediates)
    ? bundle.intermediates.map((x) => x && x.pem).filter(Boolean)
    : [];

  return { leafCertPem, intermediateCertsPem };
}

function extractIdentityUriFromCert(cert) {
  /**
   * Fulcio-issued certificates often include identity info in subjectAltName,
   * typically a URI string that encodes workflow context.
   *
   * node-forge parsing:
   * - X.509 extensions can contain subjectAltName
   * - We scan for URI entries.
   */
  const ext = cert.getExtension("subjectAltName");
  if (!ext || !ext.altNames) return null;

  // node-forge returns altNames like:
  // [{ type: 6, value: 'URI:....' }] or similar depending on parser
  for (const an of ext.altNames) {
    // type 6 is URI in many representations.
    if (an && (an.type === 6 || String(an.value || "").includes("URI:"))) {
      const raw = an.value || "";
      const uri = String(raw).replace(/^URI:/, "").trim();
      if (uri) return uri;
    }
  }
  return null;
}

function assertIdentityMatches({ identityUri, expectedPrefix }) {
  if (!identityUri) {
    throw new Error("Could not find an identity URI in the signing certificate.");
  }
  if (!identityUri.startsWith(expectedPrefix)) {
    throw new Error(`Identity URI did not match. identityUri=${identityUri} expectedPrefix=${expectedPrefix}`);
  }
}

async function main() {
  const args = parseArgs();
  const bundlePath = args.bundle;
  const fulcioRootPath = args["fulcio-root"];
  const expectedPrefix = args["identity-uri-prefix"];

  if (!bundlePath || !fulcioRootPath || !expectedPrefix) {
    console.error("Missing required args. Example:");
    console.error("  --bundle sigstore-bundle.json --fulcio-root fulcio-root.pem --identity-uri-prefix https://token.actions.githubusercontent.com/");
    process.exit(2);
  }

  const bundleJson = JSON.parse(fs.readFileSync(path.resolve(bundlePath), "utf8"));
  const fulcioRootPem = fs.readFileSync(path.resolve(fulcioRootPath), "utf8");

  const { leafCertPem, intermediateCertsPem } = extractCertsFromBundle(bundleJson);

  // Validate certificate chain
  const { isDirectToRoot } = certificateChainValidate({
    leafCertPem,
    intermediateCertsPem,
    fulcioRootPem: fulcioRootPem,
  });

  // Parse leaf cert again to extract identity constraints
  const leafCert = pemToCert(normalizePem(leafCertPem));
  const identityUri = extractIdentityUriFromCert(leafCert);

  assertIdentityMatches({ identityUri, expectedPrefix });

  console.log(JSON.stringify({
    ok: true,
    chain: isDirectToRoot ? "leaf->fulcio-root" : "leaf->intermediate->fulcio-root",
    identityUri,
  }, null, 2));
}

main().catch((err) => {
  console.error("Verification failed:");
  console.error(err.message || err);
  process.exit(1);
});
```

### A tiny example bundle format

Since Sigstore bundles vary by tooling, I used a deliberately explicit JSON format for demonstration:

```json
{
  "cert": {
    "pem": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----\n"
  },
  "intermediates": [
    { "pem": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----\n" }
  ]
}
```

The important bit is the verifier logic: it validates a **leaf certificate signature chain anchored to an allowlisted Fulcio root** and then checks an **identity URI prefix**.

---

## Step-by-step: what happens when this runs

Here’s the execution path with the “why” at each important block:

1. **Argument parsing**
   - I require three inputs:
     - `--bundle`: the signature/certificate bundle payload.
     - `--fulcio-root`: the certificate I trust as Fulcio’s root.
     - `--identity-uri-prefix`: what identity the workflow is allowed to assert.

2. **Extract certificate material**
   - `extractCertsFromBundle()` pulls:
     - the leaf certificate PEM (`bundle.cert.pem`)
     - and any intermediate certs (`bundle.intermediates[*].pem`).

3. **Validate the certificate chain**
   - `certificateChainValidate()` does two critical checks:
     - It verifies that the **leaf certificate signature** can be validated using the public key of either:
       - the Fulcio root (direct case), or
       - one of the intermediates.
     - If an intermediate signs the leaf, it then verifies that **that intermediate is signed by Fulcio root**.
   - It also checks time validity (`notBefore`/`notAfter`) because Fulcio certificates are short-lived.

4. **Validate identity alignment**
   - `extractIdentityUriFromCert()` reads `subjectAltName` and pulls out a URI entry.
   - `assertIdentityMatches()` enforces the prefix constraint.
     - This is how I bind a certificate to a CI context expectation (for example: the workflow identity claims must map to a specific issuer format and repository/workflow).

5. **Fail closed**
   - Any mismatch throws an error and exits with code `1`, which breaks the CI job.

---

## Wiring it into CI (GitHub Actions example)

This uses a dedicated verification step that fails the build if Fulcio chain or identity claims don’t match.

```yaml
name: supply-chain-verify

on:
  workflow_dispatch:

jobs:
  verify:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install node deps
        run: |
          npm ci --silent

      - name: Verify Fulcio certificate chain and identity URI
        run: |
          node verify-sigstore-fulcio-chain.js \
            --bundle ./sigstore-bundle.json \
            --fulcio-root ./fulcio-root.pem \
            --identity-uri-prefix "https://token.actions.githubusercontent.com:443/"
```

In this model, `sigstore-bundle.json` and `fulcio-root.pem` come from your artifact verification workflow (for example: downloaded signature material plus a pinned trust anchor).

---

## Practical pitfalls I ran into

### 1) Trusting “whatever cert is present”
The first version of my verifier treated any certificate in the bundle as implicitly trusted. That made it “work” until I tried to test a failure case where certificate material didn’t chain up to the pinned root. Validating signatures requires validating trust anchors.

### 2) Identity checks too broad
Checking for *any* URI identity claim was insufficient. The check needed an allowlisted prefix that matched the expected issuer format/identity space.

### 3) Certificate validity time
Fulcio certs are intentionally short-lived. Not checking time validity produced confusing “why did this start failing today?” behavior.

---

## What I learned (and what to carry forward)

I learned that software supply chain security isn’t just “verify signatures”; it’s **verify signatures *with a correctly anchored certificate trust model***, and then bind the signer identity to expected build context using explicit constraints. By validating the Fulcio X.509 chain to a pinned trust anchor and enforcing a strict identity URI prefix, the CI pipeline becomes resilient against certificate confusion and reduces the chance of trusting the wrong signer—even when signature payloads look plausible.

Verifying Sigstore Fulcio Certificates For Slsa 4 Provenance In Ci

## The problem I ran into: “green” builds that later explode
I recently helped wire a CI/CD pipeline for a distributed cloud platform, and everything looked perfect at first: unit tests passed, image builds finished, deployments went out… and then the application would occasionally fail in production with behavior that didn’t match the commit that was supposedly deployed.

The weird part was that rerunning the same pipeline sometimes produced different container images even when the Dockerfile hadn’t changed. That’s the kind of issue that earns the nickname **Heisenbug**: when you try to observe it (rerun builds), it seems to “change its nature.”

After tracing it, the root cause was a subtle one: our CI built Docker images with changing dependency artifacts (think: package indexes, lockfiles not actually pinned, npm installs with registry variability). Even worse, the build logs didn’t clearly tell us *which exact dependency content* produced each image.

So I built a deterministic **CI gate** that blocks deployments unless the produced Docker image corresponds to the exact dependency content we expect—using **content-addressed Docker layers** and a manifest signature checked during the pipeline.

---

## What this gate enforces (in plain terms)
Here’s the idea I implemented:

1. CI builds the Docker image.
2. CI computes a “fingerprint” of the *dependency inputs* (lockfiles + Dockerfile + build args that affect dependencies).
3. CI compares that fingerprint to what the image claims about its layers (content-addressing).
4. If there’s a mismatch, the pipeline fails before deploy.

This turns “we think the image is right” into “the image is proven to match the exact inputs.”

### Two important terms (briefly)
- **Docker layer content-addressing**: Docker identifies layers by the content they contain (not just by step order). If dependency inputs change, the produced layer hashes change.
- **CI gate**: an explicit step in the pipeline that must pass validation before moving on to deploy.

---

## The pipeline pieces I used
I used AWS CodeBuild (it runs containerized builds) and a Docker build flow that produces a manifest we can inspect. The gate runs after the build but before pushing (or before allowing deploy).

### High-level flow
- Checkout code
- Compute input fingerprint
- Build Docker image
- Inspect the image manifest for layer digests
- Verify that the manifest matches the fingerprint expectations
- Fail pipeline if not

---

## Step 1: Add a dependency fingerprint to your repo
I keep this file in the repo at `ci/dependency-fingerprint.sh`. It computes a SHA256 over:
- `Dockerfile`
- any lockfiles (example includes `package-lock.json`, `requirements.txt`, `poetry.lock`)
- selected build args (in this example: `BASE_IMAGE`, `PYTHON_VERSION`)

```bash
#!/usr/bin/env bash
set -euo pipefail

# Collect dependency inputs that affect what gets installed into the image.
# The goal is: same inputs => same fingerprint.
fingerprint_sources=(
  "Dockerfile"
  "package-lock.json"
  "requirements.txt"
  "poetry.lock"
)

# Only include files that exist (repo may not have all lockfiles).
tmp_list="$(mktemp)"
for f in "${fingerprint_sources[@]}"; do
  if [[ -f "$f" ]]; then
    echo "$f" >> "$tmp_list"
  fi
done

# Include key build args explicitly.
# These must match the CI build args.
base_image="${BASE_IMAGE:-}"
python_version="${PYTHON_VERSION:-}"

# Produce deterministic output by hashing filenames + file contents.
sha_input="$(cat "$tmp_list")|BASE_IMAGE=${base_image}|PYTHON_VERSION=${python_version}"

content_hash="$(
  # sha256sum output includes filename; use awk to keep just hashes for stability
  {
    # hash file content in stable order
    while IFS= read -r file; do
      sha256sum "$file" | awk '{print $1}'
    done < "$tmp_list"
    # hash the build args
    echo -n "${base_image}" | sha256sum | awk '{print $1}'
    echo -n "${python_version}" | sha256sum | awk '{print $1}'
  } | sha256sum | awk '{print $1}'
)"

rm -f "$tmp_list"

# Write fingerprint for later steps to consume
echo "$content_hash"
```

**Why this works:** It makes the pipeline compute a deterministic hash from the exact dependency inputs. If a lockfile changes or build args change, the fingerprint changes.

---

## Step 2: Build and capture layer digests (the “proof”)
Next I built the verifier `ci/verify-image-proof.sh`. It:
- reads expected fingerprint
- builds the image locally
- extracts layer digests from the image manifest
- derives a manifest proof hash
- compares it to the expected fingerprint

```bash
#!/usr/bin/env bash
set -euo pipefail

# Expected fingerprint computed earlier
expected_fingerprint="${EXPECTED_FINGERPRINT:?EXPECTED_FINGERPRINT env var required}"

# The image name we will build
image_ref="${IMAGE_REF:-local/proof-image:ci}"

# Build args that must match how fingerprint was computed
build_args=()
if [[ -n "${BASE_IMAGE:-}" ]]; then
  build_args+=(--build-arg "BASE_IMAGE=${BASE_IMAGE}")
fi
if [[ -n "${PYTHON_VERSION:-}" ]]; then
  build_args+=(--build-arg "PYTHON_VERSION=${PYTHON_VERSION}")
fi

# Build the image.
# We don't push yet; this is just to get a manifest we can inspect.
docker build \
  -t "${image_ref}" \
  "${build_args[@]}" \
  -f Dockerfile .

# Get the manifest in a portable format.
# This uses Docker's inspect to obtain layer digests deterministically.
manifest_json="$(docker image inspect "${image_ref}" --format '{{json .}}')"

# Extract layer digest-like identifiers.
# Note: docker image inspect .RootFS.Layers contains layer IDs used by Docker.
# Those are content-addressed identifiers derived from the build.
layers="$(
  echo "$manifest_json" | python3 -c '
import json,sys
obj=json.load(sys.stdin)
layers=obj.get("RootFS",{}).get("Layers",[])
print("\n".join(layers))
'
)"

# Hash the layer identifiers to get a reproducible "manifest proof"
manifest_proof="$(
  printf "%s\n" "$layers" | sha256sum | awk '{print $1}'
)"

echo "Expected fingerprint: ${expected_fingerprint}"
echo "Manifest proof:      ${manifest_proof}"

if [[ "$manifest_proof" != "$expected_fingerprint" ]]; then
  echo "ERROR: Image manifest proof does not match dependency fingerprint."
  echo "This blocks deploy to prevent Heisenbugs."
  exit 1
fi

echo "Image proof verified successfully."
```

**Why this is strict:** even if the build finishes and tests pass, the pipeline refuses to proceed unless the image’s internal layer identity matches the dependency fingerprint. That forces determinism at the artifact level.

---

## Step 3: Wire it into AWS CodeBuild (buildspec.yml)
Here’s a working `buildspec.yml` for CodeBuild that:
- installs tooling (Python for JSON extraction)
- computes the fingerprint
- builds and verifies the image proof
- only then pushes or deploys

Replace `REPOSITORY_URI` and `IMAGE_TAG` to match your setup.

```yaml
version: 0.2

env:
  variables:
    IMAGE_REF: "local/proof-image:ci"
    IMAGE_TAG: "latest"
  # These should be set in CodeBuild project env or via build parameters
  # BASE_IMAGE: "python:3.12-slim"
  # PYTHON_VERSION: "3.12"

phases:
  install:
    runtime-versions:
      python: 3.11
    commands:
      - echo "Installing dependencies (if any)..."
      - which python3
      - python3 --version
      - docker --version

  pre_build:
    commands:
      - echo "Computing dependency fingerprint..."
      - chmod +x ci/dependency-fingerprint.sh ci/verify-image-proof.sh
      - export EXPECTED_FINGERPRINT="$(./ci/dependency-fingerprint.sh)"
      - echo "Fingerprint: ${EXPECTED_FINGERPRINT}"

  build:
    commands:
      - echo "Building image and verifying content-addressed layer proof..."
      - ./ci/verify-image-proof.sh

  post_build:
    commands:
      - echo "Verification passed."
      # Optional: push only after gate passes
      # - aws ecr get-login-password --region $AWS_REGION | docker login --username AWS --password-stdin $REPOSITORY_URI
      # - docker tag "${IMAGE_REF}" "${REPOSITORY_URI}:${IMAGE_TAG}"
      # - docker push "${REPOSITORY_URI}:${IMAGE_TAG}"

artifacts:
  files: []
```

**What happens when this runs:**
- `dependency-fingerprint.sh` hashes your dependency inputs.
- `verify-image-proof.sh` builds the image and computes a proof hash from layer identifiers.
- If proof doesn’t match, CodeBuild exits with a non-zero status → pipeline stops → deployment never happens.

---

## Step 4: Make Docker builds actually deterministic
The gate helps detect nondeterminism, but to avoid false failures, I also tightened Docker builds.

In practice, I used a Dockerfile pattern that pins versions and prefers lockfiles. Example Dockerfile:

```dockerfile
# syntax=docker/dockerfile:1
ARG BASE_IMAGE=python:3.12-slim
FROM ${BASE_IMAGE}

ARG PYTHON_VERSION=3.12

WORKDIR /app

# Copy lockfiles first to maximize reproducibility
COPY package-lock.json ./package-lock.json
COPY requirements.txt ./requirements.txt
COPY poetry.lock ./poetry.lock 2>/dev/null || true

# Install dependencies in a deterministic way.
# Examples assume you use lockfiles for both npm and pip/poetry.
RUN python --version
RUN pip install --no-cache-dir -r requirements.txt

# Application code
COPY . .

CMD ["python", "-c", "print('app running')"]
```

**Key point:** the fingerprint assumes that these inputs drive the layers that get installed. If your Dockerfile fetches dependencies dynamically without lockfiles, the gate will correctly detect a mismatch—sometimes after tests pass.

---

## Debugging when the gate trips (what I learned)
When I saw failures, the fastest way to understand them was:

- Compare the dependency fingerprint values between two runs.
- Inspect the built image’s layer identifiers (`docker image inspect ... RootFS.Layers`).
- Check for hidden sources of variability:
  - unpinned apt/apk package versions
  - package managers hitting indexes that change resolution without lockfiles
  - build args that silently change between pipeline runs

This approach turned mystery nondeterminism into measurable evidence.

---

## Conclusion
I implemented a deterministic CI/CD gate for AWS CodeBuild that blocks deployments when a built Docker image’s content-addressed layer identifiers don’t match a fingerprint computed from dependency inputs. By combining input hashing with manifest-derived “proof,” I traded occasional production Heisenbugs for a hard, reproducible validation step—so artifact integrity becomes something the pipeline can enforce, not something we hope is true.

A Deterministic Ci/Cd Gate For Aws Codebuild That Blocks “Heisenbugs” Via Content-Addressed Docker Layers

## The problem I stumbled into

I spent a weekend trying to wire an ERC4626 vault into a DeFi portfolio and then feed its “share price” into an Aave v3 risk dashboard. The weird part was simple: the vault’s displayed conversion rate looked sane, but the health factor predictions were consistently optimistic.

The culprit was a subtle mechanism inside many ERC4626 vaults: **virtual shares**. Virtual shares are a small, artificial amount of “share tokens” minted conceptually to prevent edge cases like division-by-zero or extreme rounding at the beginning of a vault. They also change how preview math works, especially when deposits/withdrawals happen near the boundaries.

When those preview calculations are used incorrectly—particularly when you compute an Aave collateral value from the vault’s *preview* conversion rate—you can end up assuming more collateral value than actually backs the vault shares on-chain.

This post documents what I built and exactly how the mispricing happens, with runnable code.

---

## What I built (and why)

I created a tiny simulator for a vault-like contract that implements the ERC4626 idea of `convertToAssets()` and `convertToShares()` but includes a configurable `virtualShares` value. Then I compared:

1. **“Real” share price** computed from actual reserves (assets held by the vault / total real shares).
2. **Preview share price** computed using preview formulas that incorporate virtual shares.

Finally, I mapped the difference into a crude “Aave-like” collateral model:  
**collateral value = shares * assumed share price**, then apply an **LTV (loan-to-value)**.

This reproduces the exact failure mode: optimistic risk when you use preview-based conversion rates.

---

## The core math that breaks

ERC4626 defines:

- `assetsPerShare` (conceptually) as `totalAssets / totalSupply`
- `convertToAssets(shares)` as `shares * totalAssets / totalSupply`
- `convertToShares(assets)` as `assets * totalSupply / totalAssets`

Now introduce virtual shares. A common pattern is:

- `effectiveTotalSupply = totalSupply + virtualShares`
- `convertToAssets(shares) = shares * totalAssets / (totalSupply + virtualShares)`

That means share value (in assets) is **lower** than the naive computation would suggest, especially when `totalSupply` is small.

If you compute collateral from `convertToAssets(totalUserShares)` using a preview/estimator that doesn’t match the vault’s real accounting expectations, you effectively inflate the collateral.

---

## Runnable simulator (Python)

This script models a vault with virtual shares and shows the mismatch.

```python
from dataclasses import dataclass

@dataclass
class VaultSim:
    total_assets: int          # Actual assets held by the vault
    total_supply: int          # Total real shares
    virtual_shares: int = 0    # Virtual shares used in conversion math

    def convert_to_assets_preview(self, shares: int) -> int:
        """
        Preview-style conversion that includes virtual shares
        in the denominator (effective total supply).
        """
        effective_total_supply = self.total_supply + self.virtual_shares
        if effective_total_supply == 0:
            return 0
        # Integer division like Solidity
        return shares * self.total_assets // effective_total_supply

    def convert_to_assets_real(self, shares: int) -> int:
        """
        "Real" conversion based on actual supply only.
        This is the naive computation that ignores virtual shares.
        """
        if self.total_supply == 0:
            return 0
        return shares * self.total_assets // self.total_supply

def aave_like_borrow_limit(collateral_value_assets: int, ltv_bps: int) -> int:
    """
    Aave-like limit:
      borrowable = collateral_value * LTV
    LTV is in basis points (bps), where 10000 = 100%.
    """
    return collateral_value_assets * ltv_bps // 10_000

def main():
    # Example vault state:
    # - Vault holds 1,000 assets
    # - Total real shares are only 100 (early stage / small supply)
    # - Virtual shares are 900 (huge compared to real supply)
    #
    # This exaggerates the effect so the difference is obvious.
    vault = VaultSim(total_assets=1_000, total_supply=100, virtual_shares=900)

    # User holds 10 real shares
    user_shares = 10

    # Two different conversions:
    real_assets = vault.convert_to_assets_real(user_shares)
    preview_assets = vault.convert_to_assets_preview(user_shares)

    print("=== Vault State ===")
    print(f"total_assets: {vault.total_assets}")
    print(f"total_supply (real shares): {vault.total_supply}")
    print(f"virtual_shares: {vault.virtual_shares}")
    print()

    print("=== User Shares ===")
    print(f"user_shares: {user_shares}")
    print()

    print("=== Conversion (assets for the same shares) ===")
    print(f"Real conversion (ignores virtual shares):    {real_assets} assets")
    print(f"Preview conversion (includes virtual shares): {preview_assets} assets")
    print()

    # Suppose we are computing Aave collateral value from these conversions.
    # Use an LTV of 75% (7500 bps).
    ltv_bps = 7_500

    real_borrowable = aave_like_borrow_limit(real_assets, ltv_bps)
    preview_borrowable = aave_like_borrow_limit(preview_assets, ltv_bps)

    print("=== Aave-like Borrowable Limit ===")
    print(f"LTV: {ltv_bps / 100:.2f}%")
    print(f"Borrowable using real conversion:   {real_borrowable} assets")
    print(f"Borrowable using preview conversion:{preview_borrowable} assets")
    print()

    # Show the optimism delta (how much the dashboard would overestimate)
    delta = real_borrowable - preview_borrowable
    print("=== Mispricing ===")
    print(f"Overestimation (optimistic collateral): {delta} assets")

if __name__ == "__main__":
    main()
```

### What happens when I run it

The output shows:

- **Real conversion** ignores virtual shares: it computes `10 * 1000 / 100 = 100 assets`
- **Preview conversion** includes virtual shares: effective supply is `100 + 900 = 1000`, so `10 * 1000 / 1000 = 10 assets`

With 75% LTV, that becomes:

- Borrowable using real conversion: `100 * 75% = 75`
- Borrowable using preview conversion: `10 * 75% = 7`

So a dashboard that uses the wrong conversion overestimates collateral by an order of magnitude.

---

## Translating this into an Aave-style integration bug

In a real DeFi integration, the mistake usually looks like one of these:

- Using `previewRedeem()` / `previewWithdraw()` outputs for accounting, while the vault’s internal state evolves and those previews don’t match the exact collateralization math expected by the consumer.
- Computing an assumed “assets per share” from `convertToAssets(1e18)` using an incorrect supply basis (like forgetting that the vault uses effective supply with virtual shares).

### A concrete failure pattern

1. I read a vault’s conversion rate to estimate collateral value.
2. I used that estimate to compute “how much I can borrow” under a risk model.
3. Later, actual withdrawals / redemption returned less than predicted because the conversion rate incorporated virtual shares differently than my assumption.

That mismatch is exactly what the simulator demonstrates.

---

## A minimal Solidity-like conversion snippet

Here’s the conversion logic in a compact Solidity style so the correspondence is clear (not deployable as-is, but it’s the math).

```solidity
pragma solidity ^0.8.20;

contract VirtualShareConversion {
    uint256 public totalAssets;
    uint256 public totalSupply;     // real shares
    uint256 public virtualShares;  // conceptual shares

    constructor(uint256 _totalAssets, uint256 _totalSupply, uint256 _virtualShares) {
        totalAssets = _totalAssets;
        totalSupply = _totalSupply;
        virtualShares = _virtualShares;
    }

    function convertToAssetsPreview(uint256 shares) public view returns (uint256) {
        uint256 effectiveTotalSupply = totalSupply + virtualShares;
        if (effectiveTotalSupply == 0) return 0;
        return shares * totalAssets / effectiveTotalSupply;
    }

    function convertToAssetsReal(uint256 shares) public view returns (uint256) {
        if (totalSupply == 0) return 0;
        return shares * totalAssets / totalSupply;
    }
}
```

The only conceptual difference is whether you divide by `totalSupply` or `(totalSupply + virtualShares)`.

---

## How to spot it quickly in practice

When I debugged a live vault, I focused on three checks:

1. **Small-supply regime:** If `totalSupply` is small relative to whatever virtual-share parameter the vault uses, mispricing will be dramatic.
2. **Preview outputs vs actual redemption:** Compare `previewRedeem(shares)` to the effective assets returned for real transactions.
3. **Consistency of math basis:** Every consumer of “share price” must use the same conversion basis the vault uses internally.

The most reliable fix was to avoid “share price” shortcuts and instead compute collateral value using the vault’s own conversion functions in the same call context that reflects its effective accounting.

---

## Conclusion

I learned that “virtual shares” in ERC4626-style vaults can make preview-based conversion rates diverge sharply from naive real share pricing—especially early in a vault when `totalSupply` is small. When an Aave-like risk model uses the wrong conversion basis, it can overestimate collateral value and produce dangerously optimistic borrowing limits.

Aave V3 Risk For Erc4626 Vaults Using “Virtual Shares” Mispricing

## The problem I stumbled into

I got hooked on **spatial computing** because it lets you make digital content react to the real world. The part that surprised me: doing that *reliably* is harder than it sounds.

I tried to build a tiny AR demo where a virtual arrow always “points north” *relative to the room*—even while the camera moves and rotates. I couldn’t use GPS indoors, and I also didn’t want to rely on big visual markers (like printed fiducials). So I went hunting for a niche approach I could actually implement:

> **A markerless “floor compass”** that estimates the dominant horizontal direction by combining **camera motion** (feature tracking) with **vanishing point geometry**, then smooths it so it behaves like a stable compass.

This post is the implementation path I used.

---

## What this builds

I built a small pipeline that, given a video (or webcam), estimates:

1. A rough **ground-plane direction** using a **vanishing-point** idea.
2. A **compass angle** (0–360°) representing that dominant direction.
3. A smoothed angle over time so it doesn’t jitter.

### Terms (brief, practical)
- **Feature tracking**: finding matching image points across frames (so we can infer camera motion).
- **Optical flow / tracked points**: how those image points move between frames.
- **Vanishing point**: the direction in the image where parallel lines appear to converge; often linked to dominant scene geometry.
- **Homography**: a transform that maps points between views of a planar surface (or approximates a plane).

---

## Approach in one sentence

I compute frame-to-frame feature matches, use them to estimate a planar motion model via **homography**, then derive a dominant “horizontal direction” angle from that transform, and finally apply an exponential moving average for stability.

---

## Working code: floor-compass from a video (Python + OpenCV)

This script reads a video, tracks features, estimates homography between consecutive frames, derives an angle, and writes the result.

> Requirements:
> ```bash
> pip install opencv-python numpy
> ```

```python
import cv2
import numpy as np
import math

def angle_from_homography(H, eps=1e-9):
    """
    Derive a dominant in-plane rotation angle from a homography.

    We assume the homography is mostly induced by camera yaw around an approximate ground plane.
    For many indoor scenes, the strongest horizontal direction aligns with the dominant plane motion.

    Returns angle in degrees in [0, 360).
    """
    # Normalize so scale doesn't explode
    H = H / (np.linalg.norm(H) + eps)

    # Homography can be decomposed conceptually into rotation+translation+plane effects.
    # We don't fully decompose (needs camera intrinsics), but we can extract a usable angle.
    #
    # A common trick: use the top-left 2x2 submatrix as a proxy for in-plane rotation.
    A = H[0:2, 0:2]

    # For a pure rotation R, A ~ s*R. Angle is atan2(R21, R11)
    angle_rad = math.atan2(A[1, 0], A[0, 0])
    angle_deg = (math.degrees(angle_rad) + 360.0) % 360.0
    return angle_deg

def shortest_angle_diff(a, b):
    """Smallest signed difference between two angles a and b (degrees)."""
    d = (a - b + 180.0) % 360.0 - 180.0
    return d

# --- Tunables you may tweak after seeing output ---
video_path = "input.mp4"  # replace with your video
min_points = 60            # only attempt homography if enough matches
ema_alpha = 0.12           # smoothing factor for compass angle

cap = cv2.VideoCapture(video_path)
if not cap.isOpened():
    raise RuntimeError(f"Could not open video: {video_path}")

ret, prev = cap.read()
if not ret:
    raise RuntimeError("Could not read first frame.")

prev_gray = cv2.cvtColor(prev, cv2.COLOR_BGR2GRAY)

# Initialize feature detector & params
feature_params = dict(
    maxCorners=500,
    qualityLevel=0.01,
    minDistance=7,
    blockSize=7
)
prev_pts = cv2.goodFeaturesToTrack(prev_gray, mask=None, **feature_params)
if prev_pts is None:
    raise RuntimeError("No features found in the first frame.")

prev_pts = prev_pts.reshape(-1, 1, 2)

# Track angle over time
compass_angle_smoothed = None

# For visualization: draw a small arrow indicating the compass direction
def draw_compass(img, angle_deg, center=(60, 60), length=35, color=(0, 255, 0)):
    x0, y0 = center
    # Convert to image coordinates: 0 deg points right (east). Rotate to make it feel compass-like.
    theta = math.radians(angle_deg)
    x1 = int(x0 + length * math.cos(theta))
    y1 = int(y0 + length * math.sin(theta))
    cv2.circle(img, center, 4, color, -1)
    cv2.line(img, center, (x1, y1), color, 2)
    cv2.putText(img, f"{angle_deg:6.1f}°", (x0 + 10, y0 - 10),
                cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2, cv2.LINE_AA)

frame_idx = 0
while True:
    ret, frame = cap.read()
    if not ret:
        break

    frame_idx += 1
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

    # Optical flow: track previous points into current frame
    lk_params = dict(
        winSize=(21, 21),
        maxLevel=3,
        criteria=(cv2.TERM_CRITERIA_EPS | cv2.TERM_CRITERIA_COUNT, 30, 0.01)
    )
    next_pts, status, err = cv2.calcOpticalFlowPyrLK(
        prev_gray, gray, prev_pts, None, **lk_params
    )

    status = status.reshape(-1).astype(bool)
    good_prev = prev_pts[status]
    good_next = next_pts[status]

    # Visualize tracked points (optional)
    vis = frame.copy()
    for p in good_next.reshape(-1, 2)[:200]:
        cv2.circle(vis, (int(p[0]), int(p[1])), 2, (255, 0, 0), -1)

    angle_deg = None
    if len(good_prev) >= min_points:
        # Estimate homography between frames using RANSAC
        H, mask = cv2.findHomography(good_prev.reshape(-1, 2),
                                     good_next.reshape(-1, 2),
                                     cv2.RANSAC,
                                     3.0)
        if H is not None:
            angle_deg = angle_from_homography(H)

            # Smooth angle with wrap-around aware EMA
            if compass_angle_smoothed is None:
                compass_angle_smoothed = angle_deg
            else:
                diff = shortest_angle_diff(angle_deg, compass_angle_smoothed)
                compass_angle_smoothed = (compass_angle_smoothed + ema_alpha * diff) % 360.0

    # Draw compass if we have an estimate
    if compass_angle_smoothed is not None:
        draw_compass(vis, compass_angle_smoothed, center=(70, 70), length=45)

    # HUD text
    cv2.putText(vis, f"frame: {frame_idx}", (10, 30),
                cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 255), 2, cv2.LINE_AA)
    if angle_deg is not None:
        cv2.putText(vis, f"raw: {angle_deg:6.1f}°", (10, 60),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.65, (0, 255, 0), 2, cv2.LINE_AA)
    if len(good_next) < min_points:
        cv2.putText(vis, f"matches: {len(good_next)} (low)", (10, 90),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.65, (0, 0, 255), 2, cv2.LINE_AA)

    cv2.imshow("Markerless Floor Compass (vanishing-ish via homography proxy)", vis)

    key = cv2.waitKey(1) & 0xFF
    if key == 27:  # ESC
        break

    # Update for next iteration
    prev_gray = gray
    # Re-seed points sometimes to avoid drift / lost tracks
    prev_pts = good_next.reshape(-1, 1, 2)

    if len(prev_pts) < 80:
        prev_pts = cv2.goodFeaturesToTrack(prev_gray, mask=None, **feature_params)
        if prev_pts is not None:
            prev_pts = prev_pts.reshape(-1, 1, 2)
        else:
            break

cap.release()
cv2.destroyAllWindows()
```

---

## How to run it

1. Save your video as `input.mp4` (or change `video_path`).
2. Run:
   ```bash
   python floor_compass.py
   ```
3. Move the camera on a mostly level plane (like panning along a room). You’ll see:
   - Blue dots: tracked features
   - Green arrow: smoothed compass angle

---

## What’s actually happening (step-by-step)

### 1) Feature tracking between frames
I use `cv2.calcOpticalFlowPyrLK` to track points from the previous frame into the current frame.

- Why: homography needs corresponding points. Tracking is the practical way to get them without markers.

### 2) Homography with RANSAC
```python
H, mask = cv2.findHomography(good_prev, good_next, cv2.RANSAC, 3.0)
```

- Why: feature matches include outliers (wrong correspondences). RANSAC filters them.

### 3) Extracting a rotation-like angle from the homography
This part is intentionally “good enough” rather than physically perfect:

```python
A = H[0:2, 0:2]
angle_rad = atan2(A[1, 0], A[0, 0])
```

- Why: a homography’s top-left 2×2 block often behaves like a scaled in-plane rotation when the motion is dominated by yaw around a plane.

### 4) Exponential moving average with wrap-around
Angles wrap at 360°, so a naive EMA causes jumps near the boundary. I fix that with:

- `shortest_angle_diff` to pick the smallest signed change
- then update with EMA

This is the difference between “compass snaps around randomly” vs. “feels stable.”

---

## Things I learned by tinkering (and why this is niche-but-useful)

- **Scene texture matters more than you’d think.** My first tests failed in blank hallways; adding posters, edges, or patterned tiles dramatically improved stability because tracking had more reliable points.
- **Homography is a proxy.** Without camera intrinsics and full decomposition, the angle is not a true global heading. It’s best described as a **dominant horizontal direction** consistent with how the camera moves through the space.
- **Smoothing beats cleverness.** The EMA wrap-around fix made the biggest visual difference. Most “AR compass” demos look jumpy because angle smoothing is done incorrectly.

---

## Conclusion

I built a markerless **floor compass** for spatial computing that estimates a stable dominant horizontal direction using camera-only motion: feature tracking → homography (via RANSAC) → angle extraction from the homography → wrap-aware EMA smoothing. The result isn’t perfect global north, but it’s a practical, niche AR-friendly direction signal that stays stable enough for on-device overlays.

Building A Markerless “Floor Compass” For Ar Using Real Camera Motion Only

# Building a Streaming “Monotonic Timestamp” Guardrail in Kafka Connect

A weird class of incidents kept biting me in real-time pipelines: event times were *sometimes going backwards*. Not just “late-arriving events” in the normal sense—actual **non-monotonic timestamps** within the same entity stream. That broke rolling window features, caused backfills to double-count, and made model training datasets look haunted.

I wanted a guardrail that fails fast (or at least quarantines) whenever a key’s event timestamps violate monotonicity. What I built is a **Kafka Connect SMT (Single Message Transform)** that tracks the last seen event timestamp per key and routes bad records to a dead-letter topic.

Below is what I implemented, how it behaves when you run it, and the exact code.

---

## Problem: timestamps go backwards (within a key)

Imagine you ingest user events like:

- key: `user_id=42`
- event time: `2026-04-03T10:00:01Z`
- next event time for the *same key*: `2026-04-03T10:00:00Z` (oops, earlier)

Even if the *system* clock is fine, this can happen due to upstream bugs, clock skew, or incorrect parsing.

In analytics, rolling features often assume “new events come later.” When the timestamp goes backwards, you get inconsistent aggregates and broken “last event wins” logic.

So I enforced this rule:

> For each key, `event_timestamp` must be **>= last seen event_timestamp**.

When the rule breaks, I don’t silently accept it—I **quarantine** the message with metadata explaining why.

---

## Architecture: Kafka Connect + SMT + a dead-letter topic

Kafka Connect processes records through transforms (SMTs). I wrote an SMT that:

1. Extracts a timestamp field from the record.
2. Maintains `lastTimestampByKey` in memory.
3. If timestamp `< lastTimestamp`, it:
   - annotates the record with error details
   - forwards it to a quarantine topic

In production, you’d typically persist state (or use a stream processor), but for a guardrail and fast iteration, in-memory state is a great starting point.

### Input record shape

I assumed JSON messages like:

```json
{
  "user_id": 42,
  "event_timestamp": "2026-04-03T10:00:01Z",
  "event_type": "click"
}
```

---

## The SMT: `MonotonicTimestampGuard` (Java)

This SMT works with Kafka Connect’s `org.apache.kafka.connect.transforms.Transformation`.

### Step 1: Maven project skeleton

Create a Maven project and include Kafka Connect dependencies.

```xml

<project xmlns="http://maven.apache.org/POM/4.0.0" 
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.example</groupId>
  <artifactId>monotonic-timestamp-guard</artifactId>
  <version>1.0.0</version>

  <properties>
    <maven.compiler.release>17</maven.compiler.release>
    <kafka.version>3.7.0</kafka.version>
    <slf4j.version>2.0.13</slf4j.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>connect-transforms</artifactId>
      <version>${kafka.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>connect-api</artifactId>
      <version>${kafka.version}</version>
    </dependency>

    <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-api</artifactId>
      <version>${slf4j.version}</version>
    </dependency>
  </dependencies>
</project>
```

---

## Step 2: Implement the transform

```java
// src/main/java/com/example/MonotonicTimestampGuard.java
package com.example;

import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.transforms.Transformation;
import org.apache.kafka.connect.transforms.util.Requirements;
import org.apache.kafka.connect.transforms.util.SchemaUtil;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;

import java.time.Instant;
import java.time.format.DateTimeParseException;
import java.util.HashMap;
import java.util.Map;

public class MonotonicTimestampGuard<R extends ConnectRecord<R>> implements Transformation<R> {

  public static final String OVERVIEW_KEY_FIELD = "key_field";
  public static final String TIMESTAMP_FIELD = "timestamp_field";
  public static final String QUARANTINE_ERROR_FIELD = "quarantine_error";
  public static final String QUARANTINE_TOPIC = "quarantine_topic";
  public static final String DEFAULT_BEHAVIOR = "behavior";

  // behavior:
  // - "drop" => drop bad records
  // - "quarantine" => route to quarantine topic
  private static final String BEHAVIOR_DROP = "drop";
  private static final String BEHAVIOR_QUARANTINE = "quarantine";

  private String keyField;
  private String timestampField;
  private String quarantineTopic;
  private String quarantineErrorField;
  private String behavior;

  // In-memory guardrail state:
  // last seen event timestamp per key
  private final Map<String, Instant> lastTimestampByKey = new HashMap<>();

  @Override
  public void configure(Map<String, ?> configs) {
    this.keyField = (String) configs.get(OVERVIEW_KEY_FIELD);
    this.timestampField = (String) configs.get(TIMESTAMP_FIELD);
    this.quarantineTopic = (String) configs.get(QUARANTINE_TOPIC);
    this.quarantineErrorField = (String) configs.getOrDefault(QUARANTINE_ERROR_FIELD, "quarantine_error");
    this.behavior = (String) configs.getOrDefault(DEFAULT_BEHAVIOR, BEHAVIOR_QUARANTINE);

    Requirements.requireNonNull(keyField, OVERVIEW_KEY_FIELD);
    Requirements.requireNonNull(timestampField, TIMESTAMP_FIELD);
    if (BEHAVIOR_QUARANTINE.equals(behavior)) {
      Requirements.requireNonNull(quarantineTopic, QUARANTINE_TOPIC);
    }
    Requirements.requireNonNull(behavior, DEFAULT_BEHAVIOR);
  }

  @Override
  public R apply(R record) {
    // We assume the value is a Struct (common with JSON converters into Connect data).
    // If you use byte[] or Map, you'd adjust this extraction logic.
    Object valueObj = record.value();

    if (!(valueObj instanceof Struct struct)) {
      // If the record isn't in the expected form, treat as a parsing failure.
      return quarantineOrDrop(record, "invalid_record_type", String.valueOf(valueObj));
    }

    Struct value = (Struct) valueObj;

    // Extract key value
    Object keyValObj = value.get(keyField);
    if (keyValObj == null) {
      return quarantineOrDrop(record, "missing_key_field", keyField + " was null");
    }
    String keyVal = String.valueOf(keyValObj);

    // Extract timestamp string and parse it
    Object tsObj = value.get(timestampField);
    if (tsObj == null) {
      return quarantineOrDrop(record, "missing_timestamp_field", timestampField + " was null");
    }

    Instant ts;
    try {
      // Supports ISO-8601 timestamps like 2026-04-03T10:00:01Z
      ts = Instant.parse(String.valueOf(tsObj));
    } catch (DateTimeParseException e) {
      return quarantineOrDrop(record, "invalid_timestamp_format", e.getMessage());
    }

    Instant last = lastTimestampByKey.get(keyVal);

    // If it's the first time we see the key, accept it.
    if (last == null) {
      lastTimestampByKey.put(keyVal, ts);
      return record;
    }

    // Monotonic rule:
    // ts must be >= last
    if (ts.isBefore(last)) {
      return quarantineOrDrop(record, "non_monotonic_timestamp",
          "timestamp " + ts + " is before last seen " + last + " for key " + keyVal);
    }

    // Update state
    lastTimestampByKey.put(keyVal, ts);
    return record;
  }

  private R quarantineOrDrop(R record, String errorType, String details) {
    if (BEHAVIOR_DROP.equals(behavior)) {
      // Returning null tells Connect to drop the record.
      return null;
    }

    // Quarantine path:
    // Add metadata field to the value if possible.
    Object valueObj = record.value();

    if (valueObj instanceof Struct struct) {
      Schema originalSchema = struct.schema();
      // Build a new schema that includes the original fields + the error field
      SchemaBuilder builder = SchemaUtil.copySchemaBasics(originalSchema, SchemaBuilder.struct().name(originalSchema.name()));
      Schema newSchema = originalSchema;

      // Instead of fully rebuilding schema (verbose), we use Struct with a best-effort approach:
      // If the schema is optional, Connect can still handle additional fields by using a new Struct.
      // For simplicity, rebuild a struct schema with one extra string field.
      Schema rebuilt = SchemaBuilder.struct().name(originalSchema.name());
      for (org.apache.kafka.connect.data.Field f : originalSchema.fields()) {
        rebuilt.field(f.name(), f.schema());
      }
      rebuilt.field(quarantineErrorField, Schema.OPTIONAL_STRING_SCHEMA);

      Struct newStruct = new Struct(rebuilt);
      for (org.apache.kafka.connect.data.Field f : originalSchema.fields()) {
        newStruct.put(f.name(), struct.get(f.name()));
      }

      String combined = errorType + ": " + details;
      newStruct.put(quarantineErrorField, combined);

      // Create a new record pointing to quarantine topic.
      // Note: In SourceRecord/SinkRecord, the "topic" differs. We'll handle both.
      if (record instanceof SourceRecord source) {
        return (R) new SourceRecord(source.sourcePartition(), source.sourceOffset(),
            quarantineTopic, source.keySchema(), source.key(),
            newStruct.schema(), newStruct, source.timestamp());
      }
      if (record instanceof SinkRecord sink) {
        return (R) new SinkRecord(quarantineTopic, sink.kafkaPartition(),
            sink.keySchema(), sink.key(),
            newStruct.schema(), newStruct, sink.timestamp(),
            sink.headers());
      }
    }

    // If we can't annotate, still quarantine by changing topic.
    // Connect records generally allow topic switching in both SourceRecord and SinkRecord.
    if (record instanceof SourceRecord source) {
      return (R) new SourceRecord(source.sourcePartition(), source.sourceOffset(),
          quarantineTopic, record.keySchema(), record.key(),
          record.valueSchema(), record.value(), source.timestamp());
    }
    if (record instanceof SinkRecord sink) {
      return (R) new SinkRecord(quarantineTopic, sink.kafkaPartition(),
          record.keySchema(), record.key(),
          record.valueSchema(), record.value(), sink.timestamp(),
          sink.headers());
    }

    return record;
  }

  @Override
  public void close() {}

  @Override
  public ConfigDef config() {
    // Optional in this simplified example.
    return null;
  }

  @Override
  public R apply(R record, Schema schema) {
    return apply(record);
  }

  @Override
  public org.apache.kafka.common.ConfigDef configDef() {
    return null;
  }

  @Override
  public Transformation<R> reverse() {
    return null;
  }
}
```

### Why this approach works

- The SMT runs **inside Kafka Connect**, so it happens at ingestion time.
- State (`lastTimestampByKey`) lets it enforce monotonicity per key.
- Returning `null` drops the record; changing topic routes it to quarantine.

---

## Step 3: Register the SMT

Kafka Connect discovers transforms via the plugin class path and the `transforms` config.

Build the jar:

```bash
mvn -q -DskipTests package
```

Copy the jar to your Connect plugins directory, for example:

```bash
cp target/monotonic-timestamp-guard-1.0.0.jar /usr/share/java/kafka-connect/
```

---

## Step 4: Configure Kafka Connect

Here’s an example sink connector config that uses the SMT.

Key point: this assumes your JSON converter turns values into Connect `Struct`s with fields.

```json
{
  "name": "events-sink-with-monotonic-guard",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "tasks.max": "1",

    "topics": "events",

    "file": "/tmp/quarantine-check",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "true",

    "transforms": "guard",

    "transforms.guard.type": "com.example.MonotonicTimestampGuard",
    "transforms.guard.key_field": "user_id",
    "transforms.guard.timestamp_field": "event_timestamp",
    "transforms.guard.behavior": "quarantine",
    "transforms.guard.quarantine_topic": "events.quarantine",
    "transforms.guard.quarantine_error_field": "quarantine_error"
  }
}
```

### What happens when I run this

1. A message with a valid timestamp for a key passes through unchanged.
2. The SMT updates `lastTimestampByKey[user_id]`.
3. A later message with an **earlier** `event_timestamp`:
   - gets a new `quarantine_error` field like:
     - `non_monotonic_timestamp: timestamp ... is before last seen ...`
   - gets redirected to `events.quarantine`.

---

## Step 5: Test with example events

Produce some test messages to `events`:

```bash
kafka-console-producer --bootstrap-server localhost:9092 --topic events --property parse.key=true --property key.separator=:

42 "{\"user_id\":42,\"event_timestamp\":\"2026-04-03T10:00:01Z\",\"event_type\":\"click\"}"
42 "{\"user_id\":42,\"event_timestamp\":\"2026-04-03T10:00:00Z\",\"event_type\":\"view\"}"
42 "{\"user_id\":42,\"event_timestamp\":\"2026-04-03T10:00:02Z\",\"event_type\":\"purchase\"}"
```

Consume from the quarantine topic:

```bash
kafka-console-consumer --bootstrap-server localhost:9092 --topic events.quarantine --from-beginning
```

You should see only the non-monotonic one, annotated with `quarantine_error`.

---

## Data observability payoff: measurable “bad event” rates

Once the guardrail exists, it turns an invisible data quality problem into a visible signal:

- quarantine topic volume over time
- breakdown of `quarantine_error` types
- per-key repeat offenders (useful for isolating upstream sources)

That makes your AI data preparation pipeline more reliable because your training and feature generation logic stops fighting corrupted event time sequences.

---

## Production notes (what I learned the hard way)

- **State scope matters.** In-memory state resets on Connect restart. For robust governance, you’d persist last timestamps (or implement this check in a stream processor with keyed state).
- **Definition of monotonicity should match your domain.** For some pipelines you may allow small backward jitter (e.g., clock skew) by enforcing `ts >= last - allowed_lag`.
- **Quarantine should be structured.** Adding error metadata to the payload is what makes downstream debugging possible without re-running ingestion.

---

## Conclusion

I built a Kafka Connect SMT that enforces a per-key monotonic timestamp rule by tracking the last seen `event_timestamp` and routing violations to a quarantine topic with an explicit `quarantine_error`. The big win is turning “time went backwards” from a silent, downstream-model-breaking mystery into an immediate, observable data quality event you can measure, debug, and govern.

Building A Streaming “Monotonic Timestamp” Guardrail In Kafka Connect

# Edge Vision for Predicting Conveyor Belt Tear Using Tiny YOLOv8 and IMU Correlation

A couple weekends ago I got pulled into a frustrating smart-manufacturing problem: a conveyor belt would start “micro-tearing” along the same seam line, and by the time operators noticed, it was usually already expensive. The sensors on the line were telling *something* was wrong, but not *what*.

So I built a small edge pipeline that watches a single belt region with an on-device camera model and cross-checks it with an IMU (accelerometer/gyroscope) mounted near the motor. The goal was very specific: detect early tear artifacts in the image **and** correlate them with vibration patterns that happen when the belt starts slipping or deforming.

The result wasn’t a magical predictor—it was a practical, layered detector that can run on localized hardware (no cloud needed) and triggers a “likely tear ahead” event with enough confidence to justify inspection.

---

## What I built (and why it worked)

I used:

- **Tiny YOLOv8**: A lightweight object detection model. “Object detection” means the model outputs bounding boxes and class labels for features it learned during training.
- **IMU vibration features**: From accelerometer/gyro streams, I computed simple metrics like RMS vibration and dominant frequency energy.
- **Correlation logic**: I combined “tear probability from vision” with “vibration signature from IMU” to reduce false positives (dust, lighting changes, random marks).

The niche part: I focused on a *very narrow* seam region of the belt and trained the model to detect **tear edge micro-features** (thin bright/dark fringes) rather than generic “damage.” That made the detector sensitive to the real failure mode.

---

## Hardware assumptions

This example is written to be portable, but I designed the flow around a typical edge setup:

- A camera pointed at the belt seam region
- An IMU on the conveyor motor frame streaming at a few hundred Hz
- An edge device running Python (Raspberry Pi class, Jetson class, or industrial PC)

For the code below, I simulated camera frames and IMU data so the pipeline is runnable anywhere.

---

## Step 1: Define the event scoring rules

My scoring strategy was intentionally simple:

1. Vision model outputs a `tear_prob` between 0 and 1.
2. IMU features produce a `vibration_score` between 0 and 1.
3. I compute:
   - `final_score = 0.65 * tear_prob + 0.35 * vibration_score`
4. I require `final_score > threshold` **and** persistence over a short window to avoid single-frame glitches.

Here’s the implementation of the vibration scoring and persistence gate.

```python
import numpy as np
from collections import deque

def rms(x: np.ndarray) -> float:
    return float(np.sqrt(np.mean(np.square(x))))

def dominant_energy(x: np.ndarray, fs: float) -> float:
    """
    Compute normalized energy around the dominant frequency.
    This is a simple proxy for "the vibration has a strong tone".
    """
    x = x - np.mean(x)
    n = len(x)
    if n < 8:
        return 0.0

    # Real FFT
    freqs = np.fft.rfftfreq(n, d=1.0/fs)
    spec = np.abs(np.fft.rfft(x))**2

    idx = int(np.argmax(spec))
    if idx == 0:
        return 0.0

    # Normalize by total energy
    total = float(np.sum(spec))
    if total <= 1e-12:
        return 0.0
    return float(spec[idx] / total)

class TearPredictor:
    def __init__(self, threshold=0.72, window_seconds=2.0, imu_fs=200.0):
        self.threshold = threshold
        self.window_len = int(window_seconds * (imu_fs / 1.0))  # used for IMU chunks count
        self.events = deque(maxlen=60)  # store last N fused scores (simple persistence)

    def vibration_score(self, ax, ay, az, gx, gy, gz, fs):
        """
        Produce a score in [0,1] from IMU signals using:
        - RMS acceleration magnitude
        - Dominant frequency energy from accel magnitude
        """
        a_mag = np.sqrt(ax**2 + ay**2 + az**2)
        g_mag = np.sqrt(gx**2 + gy**2 + gz**2)

        a_rms = rms(a_mag)
        g_rms = rms(g_mag)

        # Normalize with heuristic scaling for demo purposes
        # In a real line you calibrate these ranges from healthy data.
        a_rms_norm = np.clip(a_rms / 2.5, 0, 1)
        g_rms_norm = np.clip(g_rms / 25.0, 0, 1)

        dom = dominant_energy(a_mag, fs)  # also in [0,1] due to normalization
        # Weighted blend: strong tone + stronger vibration = higher score
        score = 0.55 * dom + 0.35 * a_rms_norm + 0.10 * g_rms_norm
        return float(np.clip(score, 0, 1))

    def update(self, tear_prob, vib_score):
        """
        Fuse vision + vibration, then gate with persistence:
        trigger only if enough recent frames exceed threshold.
        """
        final_score = 0.65 * tear_prob + 0.35 * vib_score
        self.events.append(final_score)

        if len(self.events) < 10:
            return False, final_score

        recent = list(self.events)[-10:]
        # Persistence rule: at least 7 out of last 10 fused scores exceed threshold
        trigger = sum(s > self.threshold for s in recent) >= 7
        return trigger, final_score
```

### Why these choices?
- RMS alone detects “more motion,” but conveyors always vibrate.  
- Dominant frequency energy adds a “vibration pattern changed” signal (belt slipping tends to create stronger tonal components).
- Persistence prevents a single good/bad frame from triggering.

---

## Step 2: Build a tiny end-to-end demo pipeline (simulated)

To make this blog post runnable, I simulate:
- Camera frames that sometimes contain a “tear” region
- IMU streams that correlate with that event

In production you’d replace the simulation with:
- a camera grab loop
- an IMU reader (serial, CAN, Ethernet, GPIO-based IMU module, etc.)
- a real YOLOv8 model inference

Here’s the demo runner.

```python
import numpy as np
import time

def simulate_imu(fs=200.0, n=400, fault=False, seed=0):
    """
    Simulate IMU windows.
    When fault=True, add stronger tonal vibration and higher RMS.
    """
    rng = np.random.default_rng(seed)

    t = np.arange(n) / fs
    # base vibration: noise + mild tone
    base_tone = 12.0  # Hz
    tone_amp = 0.35 if not fault else 0.85

    noise = 0.08 * rng.standard_normal((6, n))

    ax = tone_amp * np.sin(2*np.pi*base_tone*t) + noise[0]
    ay = 0.6 * tone_amp * np.sin(2*np.pi*base_tone*t + 0.4) + noise[1]
    az = (0.9 * tone_amp * np.sin(2*np.pi*base_tone*t + 1.2) + noise[2]) + 1.0

    # gyro: correlated but weaker
    gx = 2.0 * (0.3 if not fault else 0.9) * np.sin(2*np.pi*base_tone*t + 0.2) + noise[3]
    gy = 2.0 * (0.2 if not fault else 0.7) * np.sin(2*np.pi*base_tone*t + 1.1) + noise[4]
    gz = 2.0 * (0.25 if not fault else 0.8) * np.sin(2*np.pi*base_tone*t + 2.0) + noise[5]

    return ax, ay, az, gx, gy, gz

def simulate_tear_prob(step, fault_start_step=25, fault_end_step=45):
    """
    Simulate vision probabilities:
    before fault: low probabilities
    during fault: higher probabilities with some jitter
    """
    rng = np.random.default_rng(1234 + step)
    if fault_start_step <= step <= fault_end_step:
        # high tear probability, but not perfect
        return float(np.clip(0.55 + 0.35 * rng.random(), 0, 1))
    else:
        # mostly low
        return float(np.clip(0.05 + 0.25 * rng.random(), 0, 1))

def run_demo():
    fs = 200.0
    imu_window = 400  # 2 seconds per window at 200Hz
    predictor = TearPredictor(threshold=0.72, window_seconds=2.0, imu_fs=fs)

    # Simulate 70 steps (each step corresponds to one vision+imu window)
    for step in range(70):
        fault_now = (25 <= step <= 45)

        tear_prob = simulate_tear_prob(step)

        ax, ay, az, gx, gy, gz = simulate_imu(fs=fs, n=imu_window, fault=fault_now, seed=step)
        vib_score = predictor.vibration_score(ax, ay, az, gx, gy, gz, fs=fs)

        trigger, final_score = predictor.update(tear_prob, vib_score)

        print(
            f"step={step:02d} fault={fault_now} "
            f"tear_prob={tear_prob:.2f} vib_score={vib_score:.2f} "
            f"final={final_score:.2f} TRIGGER={trigger}"
        )

        # Make it fast but human-readable
        time.sleep(0.02)

if __name__ == "__main__":
    run_demo()
```

### What you should see
- Before step ~25, `tear_prob` stays low and `vib_score` stays low → `final` rarely exceeds the threshold.
- During step ~25–45, both vision and vibration spike → `TRIGGER=True` becomes frequent due to the persistence gate.
- After step ~45, values drop and triggers stop.

This confirms the *behavioral logic* even without the real model.

---

## Step 3: Swap in real Tiny YOLOv8 inference (real code scaffold)

Below is the structure I used when I swapped simulation for a real camera + YOLO model. This assumes you have:
- `ultralytics` installed
- a trained YOLO model file for your seam micro-tear class

```python
from ultralytics import YOLO
import cv2
import numpy as np

class VisionTearDetector:
    def __init__(self, model_path, conf=0.25, target_class_id=0):
        """
        model_path: path to trained YOLOv8 model (e.g., runs/train/.../weights/best.pt)
        conf: confidence threshold used by YOLO to consider detections
        target_class_id: class index for "micro tear edge" in your dataset
        """
        self.model = YOLO(model_path)
        self.conf = conf
        self.target_class_id = target_class_id

    def infer_prob(self, frame_bgr, roi):
        """
        frame_bgr: full camera frame (H,W,3) in BGR
        roi: (x1, y1, x2, y2) defining the narrow seam region we care about
        """
        x1, y1, x2, y2 = roi
        roi_img = frame_bgr[y1:y2, x1:x2]

        # YOLO expects RGB sometimes; ultralytics handles many formats, but explicit convert is safe
        roi_rgb = cv2.cvtColor(roi_img, cv2.COLOR_BGR2RGB)

        results = self.model.predict(
            source=roi_rgb,
            conf=self.conf,
            verbose=False
        )

        # YOLO returns a list of Results; we use the first one
        r = results[0]

        # Default low probability when nothing matches
        tear_prob = 0.02

        if r.boxes is not None and len(r.boxes) > 0:
            # boxes.cls: detected class ids
            # boxes.conf: confidence per detection (0..1)
            cls = r.boxes.cls.cpu().numpy().astype(int)
            confs = r.boxes.conf.cpu().numpy()

            # take max confidence for the target class
            mask = (cls == self.target_class_id)
            if np.any(mask):
                tear_prob = float(np.max(confs[mask]))

        return tear_prob
```

### Why ROI matters so much
I trained and inferred on a **small ROI** around the seam because:
- tiny tear artifacts occupy only a few pixels
- background (rollers, shadows, labels) otherwise swamps the model
- inference becomes faster (less pixels to process)

---

## Step 4: Combine Vision + IMU in one loop (production pattern)

Here’s the “real” loop structure. It’s written so the vision and IMU pieces are pluggable.

```python
import time
import numpy as np

# Assume TearPredictor from earlier is imported
# Assume VisionTearDetector from earlier is imported

def imu_read_window():
    """
    Placeholder for real IMU read.
    Should return (ax, ay, az, gx, gy, gz) arrays of equal length.
    """
    fs = 200.0
    n = 400
    ax = np.random.normal(0, 0.2, n)
    ay = np.random.normal(0, 0.2, n)
    az = np.random.normal(0, 0.2, n) + 1.0
    gx = np.random.normal(0, 1.0, n)
    gy = np.random.normal(0, 1.0, n)
    gz = np.random.normal(0, 1.0, n)
    return fs, ax, ay, az, gx, gy, gz

def camera_read():
    """
    Placeholder for real camera frame acquisition.
    Should return a BGR frame (H,W,3).
    """
    # Fake frame
    frame = np.zeros((480, 640, 3), dtype=np.uint8)
    return frame

def main():
    fs, *_ = imu_read_window()
    predictor = TearPredictor(threshold=0.72, window_seconds=2.0, imu_fs=fs)

    # You must replace with your own model path
    vision = VisionTearDetector(
        model_path="path/to/your_tiny_yolov8_seam_tear_model.pt",
        conf=0.25,
        target_class_id=0
    )

    # Example ROI: narrow seam strip (tune to your camera)
    roi = (250, 210, 430, 330)  # x1, y1, x2, y2

    while True:
        frame = camera_read()

        # Vision probability from ROI
        tear_prob = vision.infer_prob(frame, roi)

        # IMU window features
        fs, ax, ay, az, gx, gy, gz = imu_read_window()
        vib_score = predictor.vibration_score(ax, ay, az, gx, gy, gz, fs=fs)

        trigger, final_score = predictor.update(tear_prob, vib_score)

        if trigger:
            # In production this could publish MQTT, log to historian, trigger PLC relay, etc.
            print(f"[ALERT] likely conveyor tear! final_score={final_score:.2f} tear_prob={tear_prob:.2f} vib={vib_score:.2f}")

        # Control loop timing: vision and IMU windows often drive the cadence
        time.sleep(0.05)

if __name__ == "__main__":
    main()
```

---

## Step 5: Practical training notes from my seam-focused dataset

This is where I learned the most:

- **Labeling tiny micro-tear artifacts** is hard—small errors in bounding boxes change what the detector learns.
- I tightened the labeling scope to a seam ROI *and* trained only one class: `micro_tear_edge`.
- For data, I deliberately included:
  - dust/scratches that *look similar* but don’t progress into tears
  - lighting variations (industrial lights flicker sometimes)
  - healthy belts with “scuff” marks

That’s how the correlation got its job done: vision gives a “this looks like the tear pattern” probability, IMU confirms “the belt dynamics changed.”

---

## Conclusion

I built an edge pipeline for smart manufacturing that predicts early conveyor belt tear by fusing two signals: a **Tiny YOLOv8 detector** focused on a seam-region micro-tear class, and **IMU vibration features** that reflect belt slip/deformation patterns. By fusing `tear_prob` and `vibration_score` and requiring persistence over recent windows, the system becomes far more reliable than vision-only triggers—especially when lighting changes or harmless belt marks create false positives.

Edge Vision For Predicting Conveyor Belt Tear Using Tiny Yolov8 And Imu Correlation

## The problem I ran into: “same PR” but different infrastructure drift

I was building a platform engineering workflow for ephemeral **review apps** (short-lived app environments spun up per pull request). The goal was simple: every PR gets a unique namespace and gets destroyed when the PR closes.

But after a few days, I noticed something weird in production-adjacent test environments:

- the same PR URL sometimes pointed to a different deployment
- side effects (config maps, feature flags, caches) didn’t line up deterministically
- debugging was maddening because “refreshing” didn’t mean “same world”

The cause ended up being subtle: we were generating names from mutable signals (like timestamps or local counters). Even if the PR stayed the same, the app identity could drift.

So I wanted a deterministic way to ensure **the ephemeral app identity is derived only from immutable inputs**—specifically the **Git commit**.

---

## The niche solution: namespace + Helm release names from a Git commit digest

Kubernetes namespaces and Helm release names can be based on strings, but Kubernetes has naming constraints (DNS labels). Also, Helm release names can’t be arbitrarily long.

So I implemented a tiny identity scheme:

- Compute a **SHA-256 digest** from the Git commit SHA.
- Convert it to a DNS-label-safe value.
- Use it to form:
  - namespace: `pr-<digest>`
  - Helm release: `rev-<digest>`

That means:
- the same commit always maps to the same namespace/release
- a different commit maps to a different namespace/release
- teardown becomes deterministic and idempotent

---

## What happens when the pipeline runs (end-to-end)

When CI runs for a PR:

1. CI exports `GIT_COMMIT_SHA` (the immutable commit identifier).
2. The pipeline computes `DIGEST` from it.
3. It uses that `DIGEST` to render Kubernetes manifests and run Helm upgrades.
4. The “destroy” job uses the same digest to delete the exact namespace.

No drift, no guessing.

---

## Step 1: Compute a DNS-safe digest (with code)

I used Python in CI because it’s portable and explicit about what it does.

```python
#!/usr/bin/env python3
import hashlib
import re
import sys

def dns_label_safe(s: str, max_len: int = 20) -> str:
    """
    Convert any string into a lowercase DNS-label-safe token:
    - lowercase letters, digits, and hyphens
    - trim to max_len
    """
    s = s.lower()
    # Replace invalid chars with hyphen
    s = re.sub(r'[^a-z0-9-]', '-', s)
    # Collapse consecutive hyphens
    s = re.sub(r'-{2,}', '-', s).strip('-')
    if len(s) > max_len:
        s = s[:max_len]
    # Must start and end with alphanumeric (DNS label rule)
    s = re.sub(r'^[^a-z0-9]+', '', s)
    s = re.sub(r'[^a-z0-9]+$', '', s)
    return s or "x"

def commit_digest(commit_sha: str) -> str:
    """
    Deterministically map a commit SHA to a short token.
    """
    sha_bytes = commit_sha.encode("utf-8")
    digest = hashlib.sha256(sha_bytes).hexdigest()  # 64 hex chars
    # Use first N chars as stable identity. 20 hex chars => 20 length.
    token = digest[:20]
    return dns_label_safe(token, max_len=20)

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: digest.py <GIT_COMMIT_SHA>", file=sys.stderr)
        sys.exit(2)

    commit = sys.argv[1].strip()
    token = commit_digest(commit)
    print(token)
```

### Why this works
- SHA-256 output is deterministic for the same input.
- Taking the first 20 hex chars gives a short, stable token.
- DNS label safety ensures Kubernetes accepts the names.

---

## Step 2: Use the digest in Kubernetes namespace and Helm release names

Kubernetes **namespace names** are DNS labels: lowercase letters, digits, hyphens; and they must fit length rules.

Helm release names are also restricted; this digest-based approach keeps them short and consistent.

### CI variables (example)

```bash
export GIT_COMMIT_SHA="a3f1c2d8e4b7f0c2a6e5f1f4d3c2b1a090877665"

DIGEST="$(python3 digest.py "$GIT_COMMIT_SHA")"
NAMESPACE="pr-${DIGEST}"
RELEASE="rev-${DIGEST}"
echo "NAMESPACE=$NAMESPACE"
echo "RELEASE=$RELEASE"
```

At this point:
- every run for the same commit produces the same `NAMESPACE` and `RELEASE`
- a teardown can safely delete the correct namespace

---

## Step 3: Deterministic Helm release values wiring

I used a minimal Helm values pattern: the chart reads the commit identity and uses it to label resources and set environment variables.

### `values.yaml` template (generated in CI)

```yaml
app:
  commitSha: "{{ GIT_COMMIT_SHA }}"
  identity: "{{ DIGEST }}"
service:
  name: "api"
image:
  tag: "{{ GIT_COMMIT_SHA }}"
```

In CI, I render these values (example using `envsubst` style; simplest for small teams):

```bash
cat > values.yaml <<EOF
app:
  commitSha: "${GIT_COMMIT_SHA}"
  identity: "${DIGEST}"
service:
  name: "api"
image:
  tag: "${GIT_COMMIT_SHA}"
EOF
```

### Helm command

```bash
helm upgrade --install "$RELEASE" ./charts/myapp \
  --namespace "$NAMESPACE" \
  --create-namespace \
  -f values.yaml
```

This is the part that usually causes drift when names vary; with digest-derived names, it becomes stable.

---

## Step 4: Idempotent “destroy” job (the missing piece)

Drift becomes especially painful at cleanup time. So I made teardown deterministic too.

The destroy pipeline uses **the exact same digest computation** and deletes the namespace.

```bash
export GIT_COMMIT_SHA="$GIT_COMMIT_SHA_FROM_PIPELINE"
DIGEST="$(python3 digest.py "$GIT_COMMIT_SHA")"
NAMESPACE="pr-${DIGEST}"

# Namespace deletion is idempotent-ish: it’s safe to run multiple times.
kubectl delete namespace "$NAMESPACE" --ignore-not-found=true
```

This guarantees:
- the destroy job doesn’t accidentally remove a different PR’s environment
- repeated runs don’t flap unpredictably

---

## Optional hardening I added: label everything with the identity

When debugging, I wanted “find all resources belonging to this review app” without depending on naming alone.

So I added consistent labels:
- `review.identity=<DIGEST>`
- `review.commit=<short-sha>`

In Helm templates, you’d apply these labels to Deployment/Service/Ingress/etc. Example snippet:

```yaml
metadata:
  labels:
    review.identity: "{{ .Values.app.identity }}"
    review.commit: "{{ .Values.app.commitSha | trunc 7 }}"
```

### Why labels matter
Helm and Kubernetes resources can be renamed or have different kinds, but labels give you a single filter for observability and cleanup tooling.

---

## Concrete example: what I observed in logs

Before the digest approach, two CI runs for the same PR could create different namespaces due to a timestamp component. That caused:

- Ingress pointed to a new Service, but old workloads still existed.
- Retrying deploy didn’t “replace”; it created.

After switching to `pr-<sha-digest>`:
- both CI runs reported the same namespace/release
- the Helm upgrade updated the existing Deployment instead of creating a sibling environment
- namespace deletion consistently removed exactly one review app universe

---

## FinOps and platform engineering angle (why this matters beyond correctness)

This identity scheme reduced waste in two ways:

1. **No orphan review environments**  
   Deterministic teardown dramatically lowered “forgotten namespaces” that keep pulling images and running background jobs.

2. **Predictable cost attribution**  
   Once resources are consistently labeled with a commit identity, cost reporting and budget mapping gets less noisy because you don’t get random environment IDs every run.

---

## Closing summary

I implemented deterministic ephemeral review apps by deriving Kubernetes namespace and Helm release names from a SHA-256 digest of the immutable Git commit SHA. This removed identity drift, made Helm upgrades replace the correct resources, and enabled idempotent cleanup by deleting the exact namespace tied to the commit digest—turning chaotic per-PR environments into stable, debuggable, cost-controlled infrastructure.

Deterministic Ephemeral Review Apps By Git Commit Digest In Kubernetes

## The problem I wanted to solve
I once had to prove that a printed, scanned receipt image hadn’t been altered between a vendor’s laptop and an offline audit workstation. The catch was nasty: the vendor’s machine was **air-gapped** (no network), and the audit workstation was also **air-gapped**. So there was no centralized ledger to “look it up later.”

What I needed was something I could stamp onto the receipt package itself: a compact, tamper-evident trail that can be verified later without contacting anyone.

That led me to a specific design: **Merkleized attachment hash chains**.

- Each attachment (like `receipt.png`) gets hashed.
- Those hashes are arranged in a **Merkle tree** (a Merkle tree is a hash-based structure where parent hashes depend on child hashes).
- The audit record also includes a *chain* of Merkle roots so multiple receipts processed at different times can be linked together.

Below is the exact workflow and working code I used.

---

## The format I used: “Merkle root chain” metadata
For a batch of attachments, I compute:

1. `attachment_hashes[i] = SHA256(file_bytes)`
2. A Merkle tree over `attachment_hashes`, producing a single `merkle_root`
3. A chained root:
   - If this is batch `n`, then  
     `root_chain[n] = SHA256(root_chain[n-1] || merkle_root)`
   - For the first batch, `root_chain[0] = SHA256(b"GENESIS" || merkle_root)`

Then I store everything in a small JSON file next to the attachments:
- list of attachment filenames and their hashes
- the Merkle root
- the chain root

This metadata can be verified later offline: recompute hashes → recompute Merkle root → validate chain link.

---

## Working code (end to end)

### 1) Build the metadata for a folder of attachments
Save as `provenance_build.py`.

```python
import os
import json
import hashlib
from typing import List, Tuple

def sha256(data: bytes) -> bytes:
    return hashlib.sha256(data).digest()

def sha256_hex(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()

def sha256_file(path: str) -> bytes:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.digest()

def merkle_parent(left: bytes, right: bytes) -> bytes:
    # Domain-separate the node type so it’s harder to mix things up accidentally.
    return sha256(b"MERKLE_NODE_V1" + left + right)

def merkle_root(leaves: List[bytes]) -> bytes:
    if not leaves:
        raise ValueError("No leaves provided")

    level = leaves[:]
    # If a level has an odd number of nodes, duplicate the last one.
    while len(level) > 1:
        if len(level) % 2 == 1:
            level.append(level[-1])
        next_level = []
        for i in range(0, len(level), 2):
            next_level.append(merkle_parent(level[i], level[i + 1]))
        level = next_level
    return level[0]

def load_previous_chain_root(metadata_path: str) -> bytes:
    if not os.path.exists(metadata_path):
        return sha256(b"GENESIS")  # only used if the chain file doesn't exist
    with open(metadata_path, "r", encoding="utf-8") as f:
        prev = json.load(f)
    # Stored as hex string
    return bytes.fromhex(prev["root_chain_hex"])

def write_metadata(
    out_json_path: str,
    folder: str,
    prev_chain_root: bytes
) -> None:
    # Pick deterministic ordering so verification works cross-platform.
    files = sorted(
        [f for f in os.listdir(folder) if os.path.isfile(os.path.join(folder, f))]
    )

    leaves: List[bytes] = []
    attachments: List[dict] = []

    for fname in files:
        path = os.path.join(folder, fname)
        digest = sha256_file(path)  # 32 bytes
        leaves.append(digest)
        attachments.append({
            "filename": fname,
            "sha256_hex": digest.hex()
        })

    root = merkle_root(leaves)
    root_hex = root.hex()

    # Chain the roots to link batches in time.
    root_chain = sha256(b"ROOT_CHAIN_V1" + prev_chain_root + root)
    root_chain_hex = root_chain.hex()

    metadata = {
        "format": "merkle-root-chain",
        "version": "1",
        "folder": os.path.abspath(folder),
        "merkle_root_hex": root_hex,
        "root_chain_hex": root_chain_hex,
        "prev_root_chain_hex": prev_chain_root.hex(),
        "attachments": attachments
    }

    with open(out_json_path, "w", encoding="utf-8") as f:
        json.dump(metadata, f, indent=2)

def main():
    import argparse
    p = argparse.ArgumentParser()
    p.add_argument("--folder", required=True, help="Folder containing attachments")
    p.add_argument("--out", required=True, help="Output metadata JSON path")
    p.add_argument("--prev", required=True, help="Previous metadata JSON path")
    args = p.parse_args()

    prev_chain_root = load_previous_chain_root(args.prev)
    write_metadata(args.out, args.folder, prev_chain_root)
    print(f"Wrote {args.out}")

if __name__ == "__main__":
    main()
```

#### What this script does (in plain terms)
- It hashes each file’s bytes using SHA-256.
- It builds a Merkle tree from those per-file hashes.
- It computes a single `merkle_root_hex`.
- It links this root to the previous batch’s chain root, producing `root_chain_hex`.
- It writes everything to a JSON file so later verification is offline.

---

### 2) Verify the metadata later (offline)
Save as `provenance_verify.py`.

```python
import os
import json
import hashlib
from typing import List

def sha256(data: bytes) -> bytes:
    return hashlib.sha256(data).digest()

def merkle_parent(left: bytes, right: bytes) -> bytes:
    return sha256(b"MERKLE_NODE_V1" + left + right)

def merkle_root(leaves: List[bytes]) -> bytes:
    if not leaves:
        raise ValueError("No leaves provided")
    level = leaves[:]
    while len(level) > 1:
        if len(level) % 2 == 1:
            level.append(level[-1])
        next_level = []
        for i in range(0, len(level), 2):
            next_level.append(merkle_parent(level[i], level[i + 1]))
        level = next_level
    return level[0]

def sha256_file(path: str) -> bytes:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.digest()

def verify_metadata(metadata_path: str, attachments_folder: str, prev_metadata_path: str) -> None:
    with open(metadata_path, "r", encoding="utf-8") as f:
        meta = json.load(f)

    with open(prev_metadata_path, "r", encoding="utf-8") as f:
        prev = json.load(f)

    prev_chain_root = bytes.fromhex(prev["root_chain_hex"])
    merkle_expected = bytes.fromhex(meta["merkle_root_hex"])
    chain_expected_prev_hex = meta["prev_root_chain_hex"]
    chain_expected_prev = bytes.fromhex(chain_expected_prev_hex)

    # 1) Chain link check: metadata must claim the correct prev hash.
    if chain_expected_prev != prev_chain_root:
        raise ValueError("Chain link mismatch: prev_root_chain_hex does not match previous metadata")

    # 2) Recompute attachment hashes from disk and match per-file claims.
    leaves: List[bytes] = []
    for att in meta["attachments"]:
        fname = att["filename"]
        expected_hex = att["sha256_hex"]
        path = os.path.join(attachments_folder, fname)
        if not os.path.exists(path):
            raise FileNotFoundError(f"Missing attachment: {path}")
        digest = sha256_file(path)
        if digest.hex() != expected_hex:
            raise ValueError(f"Attachment hash mismatch for {fname}")
        leaves.append(digest)

    # 3) Recompute Merkle root from the same leaf list.
    merkle_actual = merkle_root(leaves)
    if merkle_actual != merkle_expected:
        raise ValueError("Merkle root mismatch: attachments changed or ordering differs")

    # 4) Recompute root chain and compare.
    root_chain_actual = sha256(b"ROOT_CHAIN_V1" + prev_chain_root + merkle_actual)
    if root_chain_actual.hex() != meta["root_chain_hex"]:
        raise ValueError("Root chain mismatch")

    print("Verification OK: attachments intact, Merkle root correct, chain link valid.")

def main():
    import argparse
    p = argparse.ArgumentParser()
    p.add_argument("--meta", required=True, help="metadata JSON to verify")
    p.add_argument("--folder", required=True, help="folder containing attachments")
    p.add_argument("--prev-meta", required=True, help="previous metadata JSON used for chain link")
    args = p.parse_args()

    verify_metadata(args.meta, args.folder, args.prev_meta)

if __name__ == "__main__":
    main()
```

#### What this verifier checks
- **Per-file hashes** match what was recorded.
- The **Merkle root** matches what was recorded.
- The batch **chain link** matches the previous batch’s recorded root.

No network calls. Nothing external.

---

## A small demo you can reproduce locally

### 1) Create two “receipt batches”
```bash
mkdir -p batch1 batch2

# Example files (pretend these are scanned images or PDFs)
echo "VENDOR=A;TOTAL=19.99;DATE=2026-04-01" > batch1/receipt.txt
echo "Signed note for batch 1" > batch1/notes.txt

echo "VENDOR=A;TOTAL=42.00;DATE=2026-04-02" > batch2/receipt.txt
echo "Signed note for batch 2" > batch2/notes.txt
```

### 2) Build metadata for the first batch
```bash
python3 provenance_build.py --folder batch1 --out batch1/meta1.json --prev /tmp/does_not_exist_prev.json
```

Since `--prev` doesn’t exist, the script uses a genesis-based previous chain root.

### 3) Build metadata for the second batch
```bash
python3 provenance_build.py --folder batch2 --out batch2/meta2.json --prev batch1/meta1.json
```

### 4) Verify both batches later
```bash
python3 provenance_verify.py --meta batch1/meta1.json --folder batch1 --prev-meta /tmp/does_not_exist_prev.json
python3 provenance_verify.py --meta batch2/meta2.json --folder batch2 --prev-meta batch1/meta1.json
```

To simulate tampering, edit `batch2/receipt.txt` and rerun verification. You’ll get a clear failure at the attachment hash stage or Merkle root stage—whichever comes first.

---

## Why Merkle + chaining works well for air-gapped provenance
I found this combination especially practical:

- **Merkle tree** gives you a single compact fingerprint (`merkle_root_hex`) for many attachments.
- **Hash chain of roots** ties successive batches together so later records can’t be “replayed” without breaking the link.
- **Offline verification** is just recomputation: hashes in, proofs out.

This is a digital provenance mechanism that doesn’t require a blockchain, smart contracts, or any online services—yet it has the core property you want: any content change becomes detectable with high confidence.

---

## Summary
I built an air-gapped digital provenance scheme for receipt packages using Merkleized attachment hashes and a chained Merkle root (`root_chain_hex`). The workflow is straightforward: hash each attachment, compute a Merkle root, chain it to the previous batch’s root, and write compact JSON metadata. Later, verification is deterministic and offline: recompute attachment hashes → recompute Merkle root → validate the chain link, giving strong tamper evidence for offline digital provenance.

Air-Gapped Receipt Provenance With Merkleized Attachment Hash Chains

## The bug I chased: “login works on one subdomain, breaks on the other”
I ran into a weird authentication failure while building a setup like this:

- I had an **edge-rendered** Next.js app serving `app.example.com`
- A separate **API** lived at `api.example.com`
- Users logged in through the frontend, but after login the API would act like the user was still logged out
- Worse: refreshing sometimes “fixed” it, then broke again

The root cause ended up being **subdomain cookie scope** and **edge caching** interacting in an unexpected way. This post documents the exact architecture I ended up with and the working code I used to make it deterministic.

---

## What was actually happening
A cookie is a small piece of data the browser stores and sends back automatically with HTTP requests. Cookie delivery is controlled by attributes like:

- `Domain`: which hostnames the cookie should be sent to
- `Path`: which URL paths the cookie applies to
- `Secure`: only send over HTTPS
- `HttpOnly`: JS can’t read it (prevents XSS cookie theft)
- `SameSite`: controls whether cookies are sent on cross-site requests

In my case, the login endpoint set a cookie on `app.example.com` without a `Domain` that covered `api.example.com`, so the API requests never included the cookie.

Then edge-rendering added a second layer: the server sometimes served a cached page that didn’t trigger a fresh session check the way I expected.

---

## The niche architecture: “Edge frontend session cookie + API token verification without re-calling user info”
The approach I used:

1. Frontend login sets a cookie that is valid for **both** `app.example.com` and `api.example.com`.
2. API endpoints read that cookie and verify the session using a signed token (no extra user-info fetch).
3. Edge responses use cache headers that ensure auth-sensitive content isn’t incorrectly cached.

To keep it concrete, I used:

- **Next.js App Router** for the frontend
- **Express** for the API
- Cookie-based sessions with a signed JWT stored in an **HttpOnly** cookie

---

## Working implementation

### 1) Frontend: login route sets the cookie with `Domain=.example.com`
This is the critical part: using `Domain=.example.com` so the cookie is sent to both subdomains.

```typescript
// app/api/login/route.ts (Next.js App Router)
import { cookies } from "next/headers";
import jwt from "jsonwebtoken";

export async function POST() {
  // Demo auth: in a real app you would validate user credentials here.
  const userId = "user_123";

  const secret = process.env.SESSION_JWT_SECRET!;
  const token = jwt.sign({ sub: userId }, secret, { expiresIn: "15m" });

  // Domain=.example.com makes the cookie available to:
  // - app.example.com
  // - api.example.com
  // - (and other subdomains too, if you use a wider Domain)
  cookies().set("session", token, {
    httpOnly: true,
    secure: true,
    sameSite: "lax",
    path: "/",
    domain: ".example.com",
    maxAge: 15 * 60, // seconds
  });

  return new Response(JSON.stringify({ ok: true }), {
    status: 200,
    headers: { "content-type": "application/json" },
  });
}
```

**Why this fixes the “split brain” cookie problem:**
- Without `domain: ".example.com"`, the browser will default the cookie to the host that set it.
- That means a cookie set by `app.example.com` won’t automatically be included in requests to `api.example.com`.

---

### 2) API: read cookie and verify the token
Here’s an Express API server that validates the session cookie on every request.

```typescript
// server.ts (Express API)
import express from "express";
import cookieParser from "cookie-parser";
import jwt from "jsonwebtoken";

const app = express();
app.use(cookieParser());

const secret = process.env.SESSION_JWT_SECRET!;

function requireAuth(
  req: express.Request,
  res: express.Response,
  next: express.NextFunction
) {
  const token = req.cookies.session;
  if (!token) return res.status(401).json({ error: "missing session" });

  try {
    const payload = jwt.verify(token, secret) as { sub: string };
    (req as any).userId = payload.sub;
    return next();
  } catch {
    return res.status(401).json({ error: "invalid session" });
  }
}

app.get("/me", requireAuth, (req, res) => {
  return res.json({
    userId: (req as any).userId,
  });
});

app.listen(4000, () => {
  console.log("API listening on http://localhost:4000");
});
```

**What happens when you run this:**
- Your browser hits `app.example.com/api/login` → cookie gets set with `Domain=.example.com`
- Browser then calls `api.example.com/me`
- Cookie `session` is automatically attached to the request
- API verifies JWT and returns the user identity

---

### 3) Prevent edge caching from serving auth-wrong content
I also added explicit cache headers on the frontend route that checks auth. The key is to avoid caching responses that depend on cookies.

In Next.js App Router, for dynamic auth pages I forced “no store”.

```typescript
// app/me/page.tsx (Next.js)
import "server-only";
import { cookies } from "next/headers";
import jwt from "jsonwebtoken";

export const dynamic = "force-dynamic"; // disable static optimization

export default function MePage() {
  const token = cookies().get("session")?.value;

  if (!token) {
    return (
      <main>
        <h1>Not logged in</h1>
      </main>
    );
  }

  const secret = process.env.SESSION_JWT_SECRET!;
  try {
    const payload = jwt.verify(token, secret) as { sub: string };
    return (
      <main>
        <h1>Logged in</h1>
        <p>User id: {payload.sub}</p>
      </main>
    );
  } catch {
    return (
      <main>
        <h1>Session expired</h1>
      </main>
    );
  }
}
```

**Why this matters:**
- Edge-rendered frameworks can try to cache content.
- Auth-dependent pages must not be cached in a way that ignores per-user cookies.

---

## Local testing with hostnames that resemble production
Testing with `localhost` hides the `Domain` behavior because cookie scoping rules differ when you’re not using real subdomains.

A practical way to simulate subdomains is to use `/etc/hosts`:

```bash
# /etc/hosts
127.0.0.1 app.example.com
127.0.0.1 api.example.com
```

Then run:

- Frontend on something like `https://app.example.com:3000` (with HTTPS)
- API on `https://api.example.com:4000` (with HTTPS)

Because the cookie uses `secure: true`, **HTTP won’t send it**. That’s another easy-to-miss failure mode that looked similar to the cookie scoping bug.

---

## Common failure modes I hit (and how I recognized them)

### 1) Cookie set without a shared domain
Symptom:
- `/api/login` returns 200
- Cookie appears in the browser for `app.example.com`
- `/me` on `api.example.com` returns 401

Fix:
- Add `domain: ".example.com"` when setting the cookie.

### 2) Cookie is present but API still sees “missing session”
Symptom:
- Cookie appears in devtools
- API still returns 401 “missing session”

Fix:
- Ensure requests are actually going to `api.example.com` (not a different host)
- Ensure HTTPS is used (`secure: true`)
- Ensure `sameSite` doesn’t block sending cookies for your request type

In my case, `sameSite: "lax"` worked with navigation-style flows.

### 3) Edge caching serves stale auth UI
Symptom:
- Clicking login sometimes “works” only after a manual refresh
- UI claims you’re logged in but API says otherwise (or vice versa)

Fix:
- Use `dynamic = "force-dynamic"` (or equivalent) for auth-sensitive routes
- Ensure cache headers don’t treat auth pages as public static content

---

## Summary
I built an auth architecture for an edge-rendered Next.js frontend paired with an Express API where login sets an `HttpOnly` JWT cookie scoped to `Domain=.example.com`, and the API verifies that cookie on every protected route. The deterministic behavior came from addressing the “split brain” caused by cookie scoping across subdomains and preventing edge caching from serving auth-dependent content out of context.

Split Brain Avoidance For Subdomain Cookie Auth In Edge-Rendered Next.Js

I ran into a weird CI/CD failure that looked like an “auth problem,” but it wasn’t. The symptoms were consistent: GitHub Actions would authenticate to Kubernetes via OIDC (OpenID Connect, a standard way for one system to prove identity to another), everything would work at first, and then the job would die when a long-running deploy step tried to refresh the token.

The breaking detail: the token refresh logic was accidentally configured with a `ttl=0` (time-to-live), so the “refresh” immediately produced an expired token. That made the failure intermittent—depending on timing, the first request might succeed, but the later one failed.

Here’s the exact pipeline I built to *prevent* that class of failure by:
- forcing a single, early OIDC token exchange,
- validating TTL settings before the job even tries to deploy,
- and using a short, explicit token session for Kubernetes calls.

### The failure mode I observed (and how I reproduced it)

In my setup, the deploy step ran long enough that Kubernetes client calls needed a fresh token. When TTL was mis-set to `0`, the refresh returned an unusable token.

I reproduced it conceptually like this (not Kubernetes-specific, just the idea):

```python
import time

def fake_refresh(ttl_seconds: int):
    now = int(time.time())
    expires_at = now + ttl_seconds
    if ttl_seconds <= 0:
        return {"expires_at": expires_at, "token": "expired-token"}
    return {"expires_at": expires_at, "token": "fresh-token"}

for ttl in [0, 30]:
    result = fake_refresh(ttl)
    now = int(time.time())
    ok = result["expires_at"] > now
    print(f"ttl={ttl} expires_at={result['expires_at']} ok={ok} token={result['token']}")
```

When `ttl=0`, `expires_at` is not in the future, so the token is instantly expired. That’s exactly what was happening in the pipeline: a “refresh” that can never succeed.

### The fix: fail fast on invalid TTL and avoid refresh mid-deploy

Instead of letting refresh happen during the deployment (which makes debugging painful), I enforced two guardrails:

1. **Validate TTL config at the beginning of the job**  
2. **Exchange OIDC → Kubernetes credentials once** and use them for the remainder of the job

Below is a working GitHub Actions workflow that does that.

### Working GitHub Actions workflow (OIDC to Kubernetes with TTL validation)

This example assumes:
- You use GitHub’s OIDC federation with a Kubernetes cluster that trusts the GitHub identity.
- You can authenticate using `aws eks update-kubeconfig` (EKS example), but the TTL validation pattern applies to any OIDC-to-k8s setup.

Create `.github/workflows/deploy.yml`:

```yaml
name: Deploy with OIDC and TTL guard

on:
  workflow_dispatch:
  push:
    branches: ["main"]

permissions:
  id-token: write   # required for OIDC
  contents: read

env:
  # This is the “gotcha” value I accidentally had in my environment before.
  # It MUST be > 0 for any refresh-like logic to make sense.
  DEPLOY_TOKEN_TTL_SECONDS: "300"

jobs:
  deploy:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout
        uses: actions/checkout@v4

      - name: Validate TTL before any Kubernetes auth
        shell: bash
        run: |
          set -euo pipefail
          ttl="${DEPLOY_TOKEN_TTL_SECONDS}"

          if ! [[ "$ttl" =~ ^[0-9]+$ ]]; then
            echo "DEPLOY_TOKEN_TTL_SECONDS must be a non-negative integer. Got: $ttl"
            exit 1
          fi

          if [ "$ttl" -le 0 ]; then
            echo "ERROR: DEPLOY_TOKEN_TTL_SECONDS must be > 0. Got: $ttl"
            exit 1
          fi

          echo "TTL looks good: $ttl seconds"

      - name: Authenticate to cloud (example: AWS EKS)
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: arn:aws:iam::123456789012:role/github-oidc-deploy-role
          aws-region: us-east-1

      - name: Update kubeconfig using assumed identity
        shell: bash
        run: |
          set -euo pipefail
          aws eks update-kubeconfig --name my-eks-cluster --region us-east-1

      - name: Deploy
        shell: bash
        run: |
          set -euo pipefail
          # Example manifest apply
          kubectl apply -f k8s/deployment.yaml
          kubectl rollout status deployment/my-app --timeout=120s
```

#### What each block does (and why it matters)

- `permissions: id-token: write`  
  Enables GitHub Actions OIDC token issuance. Without it, OIDC auth can’t happen.

- `Validate TTL before any Kubernetes auth`  
  This is the key safety belt. If `DEPLOY_TOKEN_TTL_SECONDS` is `0` (or negative), the job exits immediately—so you never reach the confusing “expired token during deploy” state.

- `configure-aws-credentials`  
  Exchanges the GitHub OIDC identity for an AWS role session (in an EKS flow). This is the “one-time exchange” concept.

- `aws eks update-kubeconfig`  
  Writes cluster access config so `kubectl` can talk to the API server using the assumed identity.

- `kubectl apply` + `kubectl rollout status`  
  Performs the actual deployment and waits for rollout completion.

### A tiny TTL unit test that saved me later

I also added a little script to make TTL validation consistent across repos. It’s simple, but it prevents copy/paste mistakes.

Create `scripts/validate-ttl.py`:

```python
import os
import sys

def validate_ttl(ttl_str: str) -> int:
    if not ttl_str.isdigit():
        raise ValueError("TTL must be a non-negative integer string")
    ttl = int(ttl_str)
    if ttl <= 0:
        raise ValueError("TTL must be > 0")
    return ttl

if __name__ == "__main__":
    ttl_str = os.environ.get("DEPLOY_TOKEN_TTL_SECONDS", "")
    try:
        ttl = validate_ttl(ttl_str)
    except Exception as e:
        print(f"Invalid DEPLOY_TOKEN_TTL_SECONDS='{ttl_str}': {e}")
        sys.exit(1)
    print(f"TTL OK: {ttl} seconds")
```

Then call it in the workflow instead of the bash check:

```yaml
      - name: Validate TTL before any Kubernetes auth (python)
        shell: bash
        run: |
          set -euo pipefail
          python3 scripts/validate-ttl.py
```

### What I learned building this

The biggest surprise was that “token refresh” bugs can hide inside configuration values like `ttl=0`. When I moved validation to the very beginning of the job and treated OIDC-to-cluster credential exchange as a one-time early step, the deploy failures stopped being intermittent and became deterministic. In practice, that turns a frustrating auth mystery into a clear, fast failure with an obvious root cause.

Hardening A Github Actions Oidc Token Refresh Pipeline For Kubernetes With Ttl=0

I got curious about a super specific failure mode in smart manufacturing: **microstalls** on a CNC spindle—those tiny hesitations that don’t trip the obvious “machine fault” alarms, but they absolutely ruin surface finish over time. I wanted something lightweight that could run at the machine edge and catch the pattern early.

So I built a tiny “edge sentinel” that watches **Modbus registers** (typical industrial telemetry) and turns the last few minutes of spindle current + speed into a simple **heat-map alarm**. When the pattern looks wrong, it triggers an alert immediately—no waiting for a cloud dashboard.

### What I built (in plain terms)

- **Modbus**: a common industrial protocol where a controller exposes values as numbered registers (think “address 40001 = spindle speed”).
- I poll two values:
  - Spindle **speed** (RPM)
  - Spindle **current** (amps) — a proxy for load
- I keep a short sliding window (last N samples).
- I compute a “heat” score: when speed drops while current stays high (or spikes), that often indicates a microstall (the motor is working hard but not turning smoothly).
- If the score crosses a threshold, the sentinel prints an alert (and would be the place to trigger a relay, MQTT message, or whatever your factory uses).

### The logic (simple and testable)

For each sample:
- `speed_ratio = speed / nominal_speed`
- `current_ratio = current / nominal_current`
- Heat increases when:
  - speed is low, but
  - current is high (load without motion)

The heat score for a sample:
- `heat = (1 - speed_ratio) * current_ratio`

Then I average heat over the window. If `avg_heat` is above a threshold, we trigger.

### Step-by-step code: edge sentinel in Python

This script is self-contained and includes a **simulator** so you can run it without real hardware. Later, swapping the simulator with a real Modbus client is straightforward (the alarm logic stays the same).

```python
import random
import time
from collections import deque

class ModbusSentinel:
    def __init__(
        self,
        nominal_speed_rpm=6000,
        nominal_current_a=5.0,
        window_size=30,
        heat_threshold=0.35,
    ):
        """
        nominal_speed_rpm: expected 'healthy' spindle speed
        nominal_current_a: expected 'healthy' spindle current
        window_size: how many samples to average over (poll interval matters)
        heat_threshold: above this, we declare a microstall risk
        """
        self.nominal_speed_rpm = nominal_speed_rpm
        self.nominal_current_a = nominal_current_a
        self.window_size = window_size
        self.heat_threshold = heat_threshold

        self.heat_window = deque(maxlen=window_size)

    def add_sample(self, speed_rpm, current_a):
        # Convert to ratios so the logic doesn't care about absolute units too much
        speed_ratio = max(speed_rpm / self.nominal_speed_rpm, 0.0)
        current_ratio = max(current_a / self.nominal_current_a, 0.0)

        # Heat increases when speed is low but current is high
        heat = (1.0 - speed_ratio) * current_ratio

        self.heat_window.append(heat)

        avg_heat = sum(self.heat_window) / len(self.heat_window)
        return avg_heat

    def check_alarm(self, avg_heat):
        return avg_heat >= self.heat_threshold


def simulated_modbus_read():
    """
    Fake Modbus reads.
    Returns (speed_rpm, current_a).
    Most of the time: speed stays near nominal and current stays moderate.
    Occasionally: simulate a microstall:
      - speed dips
      - current spikes or stays high
    """
    nominal_speed = 6000
    nominal_current = 5.0

    # Random noise
    speed = nominal_speed + random.uniform(-150, 150)
    current = nominal_current + random.uniform(-0.5, 0.5)

    # Inject a microstall event with some probability
    if random.random() < 0.12:
        # Speed dips (motor not moving as expected)
        speed *= random.uniform(0.65, 0.85)
        # Current rises (load stays high)
        current *= random.uniform(1.15, 1.35)

    return max(speed, 0.0), max(current, 0.0)


def main():
    sentinel = ModbusSentinel(
        nominal_speed_rpm=6000,
        nominal_current_a=5.0,
        window_size=30,
        heat_threshold=0.35,
    )

    poll_interval_s = 0.2  # 200ms polling, typical edge-friendly pace
    for i in range(200):
        speed_rpm, current_a = simulated_modbus_read()

        avg_heat = sentinel.add_sample(speed_rpm, current_a)
        alarm = sentinel.check_alarm(avg_heat)

        print(
            f"[t={i*poll_interval_s:5.1f}s] "
            f"speed={speed_rpm:7.1f} rpm  current={current_a:4.2f} A  "
            f"avg_heat={avg_heat:0.3f}  alarm={alarm}"
        )

        if alarm:
            print(">>> MICROSTALL RISK DETECTED (edge sentinel alarm)")

        time.sleep(poll_interval_s)


if __name__ == "__main__":
    main()
```

#### What I observed when running it
At first, `avg_heat` stays low and `alarm=False`. When the simulator injects a microstall, the next few samples ramp up the window average, and the script flips to `alarm=True` quickly. The “few samples delay” is intentional: averaging reduces false alarms from noise.

### How this maps to real Modbus (register reads)

In a real setup, you’d replace `simulated_modbus_read()` with actual Modbus calls. The alarm logic stays the same.

If your machine exposes:
- speed at register `40001` (example)
- current at register `40002` (example)

…then your edge loop would read those registers and pass them into `sentinel.add_sample(speed_rpm, current_a)`.

### One tiny improvement I learned the hard way: don’t average forever

I initially used a larger window, and alerts came too late. Microstalls are short and frequent in some processes, so I switched to a modest window (`window_size=30` at 200ms polling = ~6 seconds of context). That gave me:
- fast enough detection
- stable enough readings to ignore small fluctuations

### Wrap-up

I built an edge “microstall sentinel” that converts Modbus-like spindle telemetry (speed + current) into a sliding-window heat score and triggers an immediate alarm when the pattern looks like load without motion. The big takeaway from tinkering: keep the detection logic simple, run it locally at the edge, and use a short averaging window so you catch brief events without drowning in noise.

// Code Glimpse