lux/aml
2
0
Fork 0
mirror of https://github.com/luxfi/aml.git synced 2026-09-24 00:25:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

ci: native runners and our own registry, not GitHub's

Browse Source 
CI and release move to .hanzo/workflows on git.hanzo.ai, and .github/workflows is
removed. The old path could not have worked: lux-private/aml has zero repository
runners, zero organisation runners, and has never run a single workflow, while
release.yml required [self-hosted, linux, amd64] and stated GitHub-hosted runners
were not a permitted fallback. A tag would have queued forever against a pool that
does not exist — a release that claims artifacts nothing built.

The native pool is live and already carries the label this needs:
git-runner-0..5 register lux-build-linux-amd64 among their labels.

Images go to oci.hanzo.ai/luxfi/aml. That is the canonical name for our own OCI
registry — registry:2, S3-backed, Hanzo IAM token auth — so both the build and the
publish run on infrastructure we operate. The luxfi/ namespace stays because the
engine is a Lux product; Hanzo AML is a thin wrapper over it, the way hanzoai/kms
wraps luxfi/kms and hanzoai/mpc wraps luxfi/mpc.

The release keeps the discipline that makes a version trustworthy: build, smoke the
image before it is allowed to exist at the registry, push, then prove the pushed
manifest resolves. A tag is a receipt for an image that exists, and verifying it
after the push is what makes it one.

The smoke checks the two behaviours that have actually regressed here rather than
liveness alone — an unauthenticated read must be refused rather than served, and an
ordinary payment must come back allowed rather than carrying a saturated score,
because a monitor that alerts on everything ranks nothing.

The /v2/ in the registry path is the OCI Distribution Specification's. Every route
this product serves is /v1/, and there are sixteen of them with no /api/ prefix.

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
zeekay
2026-07-29 16:21:52 -07:00
co-authored by Hanzo Dev
parent 58bc533040
commit 97efad3afa
12 changed files with 1462 additions and 422 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
-38
View File
@@ -1,38 +0,0 @@
name: CI
on:
push:
branches:
- main
pull_request:
branches:
- main
# A runner may carry a machine-level go env that routes some module paths around
# the proxy and the checksum database. GOENV=off ignores it, so every module
# resolves through proxy.golang.org and is verified against sum.golang.org — a
# tag that moves later cannot change what this build compiles.
env:
GOENV: off
jobs:
ci:
runs-on: [self-hosted, linux, amd64]
steps:
- uses: actions/checkout@v4
- uses: actions/setup-go@v5
with:
go-version-file: go.mod
- uses: actions/setup-node@v4
with:
node-version: 22
- run: corepack enable && corepack prepare [email protected] --activate
- run: go mod download
- run: go mod verify
- run: go vet ./...
- run: go test -race -count=1 ./...
- run: go build ./...
# Builds the UI and links it with -tags embedui, the only path that
# compiles ui/embed_prod.go.
- run: make build
.github/workflows/docker.yml
-60
View File
@@ -1,60 +0,0 @@
name: Docker
# Single-arch (linux/amd64) container build on the working `lux-build` ARC label.
# Multi-arch (arm64) will return once the per-arch `lux-build-linux-arm64`
# pool is verified online (current per-arch labels did not pick up jobs).
# No QEMU, no cross-org reusable.
on:
workflow_dispatch:
push:
branches: [main, dev, test]
tags: ['v*']
permissions:
contents: read
packages: write
env:
IMAGE: ghcr.io/luxfi/aml
jobs:
build:
runs-on: [self-hosted, linux, amd64]
steps:
- uses: actions/checkout@v4
- uses: docker/setup-buildx-action@v3
- uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Compute tags
id: meta
run: |
set -e
tags=""
if [[ "${GITHUB_REF}" == refs/tags/v* ]]; then
v="${GITHUB_REF#refs/tags/}"
tags="ghcr.io/luxfi/aml:${v},ghcr.io/luxfi/aml:latest"
elif [[ "${GITHUB_REF}" == refs/heads/main ]]; then
tags="ghcr.io/luxfi/aml:main"
elif [[ "${GITHUB_REF}" == refs/heads/dev ]]; then
tags="ghcr.io/luxfi/aml:dev"
elif [[ "${GITHUB_REF}" == refs/heads/test ]]; then
tags="ghcr.io/luxfi/aml:test"
fi
echo "tags=${tags}" >> "$GITHUB_OUTPUT"
echo "version=${v:-$(git describe --tags --always)}" >> "$GITHUB_OUTPUT"
- name: Build & push
if: steps.meta.outputs.tags != ''
uses: docker/build-push-action@v6
with:
context: .
platforms: linux/amd64
push: true
tags: ${{ steps.meta.outputs.tags }}
# The build stage stamps main.version from this; without it every
# image would report "dev".
build-args: |
VERSION=${{ steps.meta.outputs.version }}
.github/workflows/release.yml
-62
View File
@@ -1,62 +0,0 @@
name: Release
# Native binary release for amld on `v*` tags.
#
# Six-platform matrix (linux/darwin/windows × amd64/arm64) via goreleaser.
# Container image is built independently by .github/workflows/docker.yml
# (native multi-arch ARC runners — NO QEMU).
#
# Runner labels: lux-build self-hosted ARC scale-sets per the org policy.
# GitHub-hosted runners are NOT a permitted fallback.
on:
push:
tags: ['v*']
workflow_dispatch:
permissions:
contents: write
packages: write
id-token: write
# A runner may carry a machine-level go env that routes some module paths around
# the proxy and the checksum database. GOENV=off ignores it, so every module
# resolves through proxy.golang.org and is verified against sum.golang.org — a
# tag that moves later cannot change what gets released.
env:
GOENV: off
jobs:
goreleaser:
runs-on: [self-hosted, linux, amd64]
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
- uses: actions/setup-go@v5
with:
go-version-file: go.mod
# goreleaser builds ui/dist in its before-hooks and links with
# -tags embedui, so the release runner needs the Node toolchain. pnpm must
# match pnpm-lock.yaml's lockfileVersion (9.0).
- uses: actions/setup-node@v4
with:
node-version: 22
- run: corepack enable && corepack prepare [email protected] --activate
- name: Install syft for SBOM
run: |
mkdir -p "$HOME/.local/bin"
curl -sSfL https://raw.githubusercontent.com/anchore/syft/main/install.sh \
| sh -s -- -b "$HOME/.local/bin"
echo "$HOME/.local/bin" >> "$GITHUB_PATH"
- uses: goreleaser/goreleaser-action@v6
with:
distribution: goreleaser
version: "~> v2"
args: release --clean
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
.hanzo/workflows/ci.yml
+100
View File
@@ -0,0 +1,100 @@
name: CI
# Runs on the native runners at git.hanzo.ai (git-runner, label
# lux-build-linux-amd64). There are no GitHub-hosted runners in this
# organisation and none is a permitted fallback: a workflow that names one
# queues forever, which reads from the outside as a test suite that never fails.
#
# Every branch, not only main. A compliance engine's tests are the evidence that
# its controls work, so they run before the merge that would rely on them.
on:
push:
branches: ["*"]
tags-ignore: ["v*"]
pull_request:
branches: ["*"]
workflow_dispatch:
concurrency:
group: ci-${{ github.ref }}
cancel-in-progress: true
jobs:
test:
runs-on: [lux-build-linux-amd64]
timeout-minutes: 30
# A runner may carry a machine-level go env that routes some module paths
# around the proxy and the checksum database. GOENV=off ignores it, so every
# module resolves through proxy.golang.org and is verified against
# sum.golang.org — a tag that moves later cannot change what was tested.
env:
GOENV: off
GOWORK: off
steps:
- uses: actions/checkout@v4
- uses: actions/setup-go@v5
with:
go-version-file: go.mod
- uses: actions/cache@v4
with:
path: |
~/.cache/go-build
~/go/pkg/mod
key: ${{ runner.os }}-go-${{ hashFiles('go.sum') }}
restore-keys: ${{ runner.os }}-go-
- name: Modules verify against the checksum database
# An AML engine's dependency graph is part of its audit surface. This
# fails on a moved tag rather than building whatever the cache holds.
run: |
go mod download
go mod verify
- name: Formatting
run: |
unformatted=$(gofmt -l pkg cmd)
if [ -n "$unformatted" ]; then
echo "::error::not gofmt-clean:"; echo "$unformatted"; exit 1
fi
- name: Vet
run: go vet ./...
- name: Test with the race detector
# Ingest is concurrent by construction — the aggregate windows and the
# behavioural model are written on the request path — so -race is the
# test run, not an extra one.
run: go test -race -count=1 ./...
- name: Build the daemon with the embedded operator surface
run: |
CGO_ENABLED=0 go build -trimpath -tags embedui -o /tmp/amld ./cmd/amld
test -x /tmp/amld
- name: It serves, and it refuses what it should
# Two facts worth a check on every commit, because both have regressed
# before: the process comes up and answers, and an unauthenticated
# request is refused rather than served somebody's tenant data.
run: |
set -euo pipefail
dir=$(mktemp -d)
AML_TENANT_HEADER=X-Org-Id \
AML_TOKEN_KEY=$(head -c 32 /dev/urandom | xxd -p -c 64) \
/tmp/amld serve --dir "$dir" --http 127.0.0.1:8791 --dev=false &
pid=$!
for i in $(seq 1 60); do
curl -sf -o /dev/null http://127.0.0.1:8791/v1/aml/health && break
sleep 1
done
curl -sf -o /dev/null http://127.0.0.1:8791/v1/aml/health \
|| { echo "::error::amld never answered /v1/aml/health"; kill $pid; exit 1; }
code=$(curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:8791/v1/aml/cases)
kill $pid
if [ "$code" != "401" ]; then
echo "::error::an unauthenticated read of /v1/aml/cases returned $code, want 401"
exit 1
fi
echo "serves, and refuses an unauthenticated read"
.hanzo/workflows/release.yml
+150
View File
@@ -0,0 +1,150 @@
name: Release
# Builds oci.hanzo.ai/luxfi/aml on the native runners (git.hanzo.ai → git-runner).
#
# A tag push vX.Y.Z is a release: the image carries the semver tags plus the sha
# tag; a main push carries only the sha tag. The image must pass the smoke below
# before it is allowed to exist at the registry, and the pushed manifest is
# proven resolvable — so a tag can never point at an image nothing built. A
# version that claims an artifact it does not have is worse than no version,
# because a deploy pin will believe it.
#
# oci.hanzo.ai is the canonical name for our own OCI registry — registry:2,
# S3-backed, Hanzo IAM token auth, with registry.hanzo.ai kept as an alias so
# existing pins keep resolving. Builds run on our runners and publish to our
# registry; nothing in this path depends on a third party being available.
#
# The /v2/ in the registry API is the OCI Distribution Specification's, not ours.
# Every route this product itself serves is /v1/.
#
# Auth: OCI_USER / OCI_TOKEN — org Actions secrets on this forge, sourced from
# KMS. The instance GITHUB_TOKEN is a Hanzo Git token and is not a registry
# credential. The luxfi/ namespace is kept because the engine is a Lux product;
# Hanzo AML is a thin wrapper over it, the way hanzoai/kms wraps luxfi/kms.
on:
push:
branches: [main]
tags: ['v*']
workflow_dispatch:
concurrency:
group: release-${{ github.ref }}
cancel-in-progress: false
jobs:
build:
runs-on: [lux-build-linux-amd64]
timeout-minutes: 45
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Compute tags
id: meta
run: |
set -euo pipefail
SHA_TAG="oci.hanzo.ai/luxfi/aml:sha-${GITHUB_SHA:0:7}"
if [[ "$GITHUB_REF" == refs/tags/v* ]]; then
VERSION="${GITHUB_REF#refs/tags/v}"
{
echo "tags<<EOF"
echo "oci.hanzo.ai/luxfi/aml:v${VERSION}"
echo "oci.hanzo.ai/luxfi/aml:${VERSION}"
echo "$SHA_TAG"
echo "EOF"
} >> "$GITHUB_OUTPUT"
echo "probe=oci.hanzo.ai/luxfi/aml:v${VERSION}" >> "$GITHUB_OUTPUT"
echo "version=${VERSION}" >> "$GITHUB_OUTPUT"
else
{
echo "tags<<EOF"
echo "$SHA_TAG"
echo "EOF"
} >> "$GITHUB_OUTPUT"
echo "probe=$SHA_TAG" >> "$GITHUB_OUTPUT"
echo "version=sha-${GITHUB_SHA:0:7}" >> "$GITHUB_OUTPUT"
fi
- uses: docker/setup-buildx-action@v3
with:
driver: docker-container
driver-opts: network=host
- uses: docker/login-action@v3
with:
registry: oci.hanzo.ai
username: ${{ secrets.OCI_USER }}
password: ${{ secrets.OCI_TOKEN }}
- name: Build (load for smoke, not yet pushed)
uses: docker/build-push-action@v6
with:
context: .
file: Dockerfile
platforms: linux/amd64
build-args: |
VERSION=${{ steps.meta.outputs.version }}
load: true
push: false
provenance: false
tags: ${{ steps.meta.outputs.tags }}
- name: Smoke — it serves, it refuses, and it does not clear a benign payment by accident
# An image that builds but cannot answer deploys and half-works. Beyond
# liveness this checks the two behaviours that have actually regressed in
# this engine: an unauthenticated read must be refused rather than
# served, and an ordinary payment must come back allowed rather than
# carrying a saturated score — a monitoring system that alerts on
# everything ranks nothing and is indistinguishable from one that alerts
# on nothing.
run: |
set -euo pipefail
img="${{ steps.meta.outputs.probe }}"
docker run --rm --entrypoint sh "$img" -c 'command -v amld'
cid=$(docker run -d \
-e AML_TENANT_HEADER=X-Org-Id \
-e AML_TOKEN_KEY="$(head -c 32 /dev/urandom | xxd -p -c 64)" \
-p 127.0.0.1:8799:8090 "$img")
trap 'docker logs "$cid" | tail -60; docker rm -f "$cid" >/dev/null 2>&1 || true' EXIT
ready=""
for i in $(seq 1 30); do
if curl -sf -o /dev/null http://127.0.0.1:8799/v1/aml/health; then ready=1; break; fi
sleep 2
done
[ -n "$ready" ] || { echo "::error::amld never answered /v1/aml/health"; exit 1; }
code=$(curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:8799/v1/aml/cases)
[ "$code" = "401" ] || { echo "::error::unauthenticated read returned $code, want 401"; exit 1; }
body=$(curl -s -X POST http://127.0.0.1:8799/v1/aml/transactions \
-H 'X-Org-Id: smoke' -H 'Content-Type: application/json' \
-d '{"id":"smoke-1","source":"api","user_id":"u1","account_id":"a1","qty":1,"notional":100,"currency":"USD","asset_class":"fiat"}')
echo "ingest: $body"
echo "$body" | grep -q '"action":"allow"' \
|| { echo "::error::a benign \$100 payment was not allowed: $body"; exit 1; }
echo "serves, refuses, and clears a benign payment"
- name: Push
run: |
set -euo pipefail
while IFS= read -r t; do
[ -n "$t" ] && docker push "$t"
done <<< "${{ steps.meta.outputs.tags }}"
- name: Verify the pushed manifest resolves
# The tag is a receipt for an image that exists. Proving it resolves
# after the push is what makes it one.
run: |
set -euo pipefail
img="${{ steps.meta.outputs.probe }}"
for i in 1 2 3 4 5 6; do
if docker manifest inspect "$img" >/dev/null 2>&1; then
echo "$img is resolvable"; exit 0
fi
echo "manifest not visible yet (attempt ${i}/6) — retrying"; sleep 5
done
echo "::error::$img not pullable after push"; exit 1
pkg/retention/base.go
+286
View File
@@ -0,0 +1,286 @@
// Copyright 2024-2026 Lux Partners Limited. All rights reserved.
// Use of this source code is governed by the Apache 2.0
// license that can be found in the LICENSE file.
package retention
import (
"encoding/base64"
"fmt"
"time"
"github.com/hanzoai/base/core"
"github.com/hanzoai/dbx"
"github.com/luxfi/aml/pkg/store"
)
// durable keeps records in Base collections, so they are still there after a
// restart. That is not a nicety: the five-year period is the obligation, and a
// ledger that starts empty every time the process starts has kept nothing
// (AMLR Art. 77).
//
// The app is a field rather than a package handle because a transaction is a
// different app over the same data — see tx.
type durable struct{ app core.App }
func (d durable) tx(fn func(store) error) error {
return d.app.RunInTransaction(func(txApp core.App) error {
return fn(durable{app: txApp})
})
}
func (d durable) read(org, id string) (Record, error) {
rows, err := records.Find(d.app, org, "id = {:id}", "", 1, dbx.Params{"id": id})
if err != nil {
return Record{}, fmt.Errorf("%w: %w", ErrStore, err)
}
if len(rows) == 0 {
return Record{}, fmt.Errorf("%w: %s", ErrNotFound, id)
}
return read(rows[0])
}
func (d durable) retained(org, identity string) (Record, error) {
rows, err := records.Find(d.app, org, fieldIdentity+" = {:identity}", "", 1, dbx.Params{"identity": identity})
if err != nil {
return Record{}, fmt.Errorf("%w: %w", ErrStore, err)
}
if len(rows) == 0 {
return Record{}, fmt.Errorf("%w: identity %s", ErrNotFound, identity)
}
return read(rows[0])
}
// insert writes the record and the party index entries that find it. Both halves
// belong to one write, so the ledger calls this inside tx: a record indexed under
// no party cannot be found by a lookback, and a lookback that cannot find it
// answers Art. 78 with "no".
func (d durable) insert(r Record) error {
row, err := records.New(d.app, r.Org)
if err != nil {
return fmt.Errorf("%w: %w", ErrStore, err)
}
row.Id = r.ID
write(row, r)
if err := d.app.Save(row); err != nil {
return fmt.Errorf("%w: retaining %s: %w", ErrStore, r.ID, err)
}
for _, p := range r.Parties {
index, err := parties.New(d.app, r.Org)
if err != nil {
return fmt.Errorf("%w: %w", ErrStore, err)
}
index.Set(fieldParty, p)
index.Set(fieldRecord, r.ID)
if err := d.app.Save(index); err != nil {
return fmt.Errorf("%w: indexing %s: %w", ErrStore, r.ID, err)
}
}
return nil
}
func (d durable) update(r Record) error {
rows, err := records.Find(d.app, r.Org, "id = {:id}", "", 1, dbx.Params{"id": r.ID})
if err != nil {
return fmt.Errorf("%w: %w", ErrStore, err)
}
if len(rows) == 0 {
return fmt.Errorf("%w: %s", ErrNotFound, r.ID)
}
write(rows[0], r)
if err := d.app.Save(rows[0]); err != nil {
return fmt.Errorf("%w: %w", ErrStore, err)
}
return nil
}
// erase destroys whole records and every index entry that referenced them. The
// index entries go first: an entry left pointing at a destroyed record is how a
// disposed record keeps being counted as evidence of a relationship.
func (d durable) erase(rs []Record) error {
for _, r := range rs {
index, err := parties.Find(d.app, r.Org, fieldRecord+" = {:record}", "", 0, dbx.Params{"record": r.ID})
if err != nil {
return fmt.Errorf("%w: %w", ErrStore, err)
}
for _, entry := range index {
if err := d.app.Delete(entry); err != nil {
return fmt.Errorf("%w: unindexing %s: %w", ErrStore, r.ID, err)
}
}
rows, err := records.Find(d.app, r.Org, "id = {:id}", "", 1, dbx.Params{"id": r.ID})
if err != nil {
return fmt.Errorf("%w: %w", ErrStore, err)
}
for _, row := range rows {
if err := d.app.Delete(row); err != nil {
return fmt.Errorf("%w: destroying %s: %w", ErrStore, r.ID, err)
}
}
}
return nil
}
func (d durable) inside(org, relationship string) ([]Record, error) {
rows, err := records.Find(d.app, org, fieldInside+" = {:inside}", fieldOccurred+",id", 0, dbx.Params{"inside": relationship})
if err != nil {
return nil, fmt.Errorf("%w: %w", ErrStore, err)
}
return all(rows)
}
// party returns the records naming a party, read through the party index. The work
// is proportional to that party's records and not to the ledger, which is what
// Art. 78's "fully and speedily" requires.
func (d durable) party(org, party string) ([]Record, error) {
index, err := parties.Find(d.app, org, fieldParty+" = {:party}", "", 0, dbx.Params{"party": party})
if err != nil {
return nil, fmt.Errorf("%w: %w", ErrStore, err)
}
out := make([]Record, 0, len(index))
for _, entry := range index {
r, err := d.read(org, entry.GetString(fieldRecord))
if err != nil {
// An index entry pointing at a record that is not there. Repairing it is
// not this read's business; disposal is what proves the indexes, and it
// reports exactly this as a failure.
continue
}
out = append(out, r)
}
return out, nil
}
// each visits an org's records of one class, oldest event first, a page at a time.
//
// It pages by the last record seen rather than by an offset. A five-year ledger is
// walked while a disposal run may be destroying from underneath it, and an offset
// page over a shrinking collection skips rows — a skipped record is one that was
// never produced to the authority that asked for it.
func (d durable) each(org string, c Class, visit func(Record) error) error {
filter := fieldOccurred + " > {:at} || (" + fieldOccurred + " = {:at} && id > {:id})"
if c != "" {
filter = fieldClass + " = {:class} && (" + filter + ")"
}
at, id := time.Time{}, ""
for {
params := dbx.Params{"at": at, "id": id, "class": string(c)}
rows, err := records.Find(d.app, org, filter, fieldOccurred+",id", batch, params)
if err != nil {
return fmt.Errorf("%w: %w", ErrStore, err)
}
if len(rows) == 0 {
return nil
}
for _, row := range rows {
r, err := read(row)
if err != nil {
return err
}
if err := visit(r); err != nil {
return err
}
}
last := rows[len(rows)-1]
at, id = last.GetDateTime(fieldOccurred).Time(), last.Id
}
}
// expired answers which records' periods have run out, in every org, from the
// expiry index. An empty expiry is a clock that has not started — an open
// relationship and everything retained inside it — and has not run out.
func (d durable) expired(now time.Time, limit int) ([]Record, error) {
filter := fieldExpiry + " != '' && " + fieldExpiry + " <= {:now}"
rows, err := records.Across(d.app, filter, "id", limit, dbx.Params{"now": now.UTC()})
if err != nil {
return nil, fmt.Errorf("%w: %w", ErrStore, err)
}
return all(rows)
}
func (d durable) count() (int, error) {
n, err := records.Count(d.app)
if err != nil {
return 0, fmt.Errorf("%w: %w", ErrStore, err)
}
return n, nil
}
// write puts a record into a row. Every field the reader reads is set here, from
// the constants both of them name, so the two cannot drift apart.
//
// The expiry column is derived from [Record.Expiry] here and nowhere else. It
// exists so disposal can ask which records have expired without reading all of
// them to find out, and deriving it in one place is what keeps the column from
// saying something the arithmetic does not.
func write(row *core.Record, r Record) {
row.Set(fieldClass, string(r.Class))
row.Set(fieldTrigger, string(r.Trigger))
row.Set(fieldParties, r.Parties)
row.Set(fieldRef, r.Ref)
row.Set(fieldNature, r.Nature)
row.Set(fieldReason, r.Reason)
row.Set(fieldInside, r.Relationship)
row.Set(fieldOccurred, r.Occurred.UTC())
row.Set(fieldEnded, r.Ended.UTC())
row.Set(fieldStart, r.Start.UTC())
row.Set(fieldExpiry, r.Expiry().UTC())
row.Set(fieldWritten, r.Written.UTC())
row.Set(fieldBody, base64.StdEncoding.EncodeToString(r.Body))
row.Set(fieldAssess, r.Assessment)
row.Set(fieldExtended, r.Extended)
row.Set(fieldIdentity, r.identity)
}
// read takes a record out of a row. A field that cannot be decoded is an error and
// never a zero value: a body that silently read as empty is a transaction that can
// no longer be reconstructed, and a record that cannot be reconstructed in full is
// one this package would rather refuse to hand over.
func read(row *core.Record) (Record, error) {
r := Record{
ID: row.Id,
Org: row.GetString(store.Org),
Class: Class(row.GetString(fieldClass)),
Trigger: Trigger(row.GetString(fieldTrigger)),
Ref: row.GetString(fieldRef),
Nature: row.GetString(fieldNature),
Reason: row.GetString(fieldReason),
Relationship: row.GetString(fieldInside),
Occurred: row.GetDateTime(fieldOccurred).Time(),
Ended: row.GetDateTime(fieldEnded).Time(),
Start: row.GetDateTime(fieldStart).Time(),
Written: row.GetDateTime(fieldWritten).Time(),
identity: row.GetString(fieldIdentity),
}
if err := row.UnmarshalJSONField(fieldParties, &r.Parties); err != nil {
return Record{}, fmt.Errorf("%w: %s parties: %w", ErrStore, row.Id, err)
}
if err := row.UnmarshalJSONField(fieldAssess, &r.Assessment); err != nil {
return Record{}, fmt.Errorf("%w: %s assessment: %w", ErrStore, row.Id, err)
}
if err := row.UnmarshalJSONField(fieldExtended, &r.Extended); err != nil {
return Record{}, fmt.Errorf("%w: %s extension: %w", ErrStore, row.Id, err)
}
if encoded := row.GetString(fieldBody); encoded != "" {
body, err := base64.StdEncoding.DecodeString(encoded)
if err != nil {
return Record{}, fmt.Errorf("%w: %s body: %w", ErrStore, row.Id, err)
}
r.Body = body
}
return r, nil
}
func all(rows []*core.Record) ([]Record, error) {
out := make([]Record, 0, len(rows))
for _, row := range rows {
r, err := read(row)
if err != nil {
return nil, err
}
out = append(out, r)
}
return out, nil
}
pkg/retention/ledger.go
+214 -262
View File
@@ -5,40 +5,30 @@
package retention
import (
"errors"
"fmt"
"slices"
"strings"
"sync"
"time"
"github.com/google/uuid"
"github.com/hanzoai/base/core"
)
// Ledger holds an org's retained records, the party index the five-year
// lookback is answered from, and the relationship index the clock cascades
// through.
type Ledger struct {
mu sync.RWMutex
records map[string]*Record
// parties maps org and party to the records that name that party. AMLR
// Art. 78 must be answered "fully and speedily", which is an index and not a
// scan: a lookback examines one party's records, never the ledger.
parties map[string][]string
// inside maps a relationship to the records retained inside it, so ending the
// relationship starts their clocks too.
inside map[string][]string
}
// Ledger holds an org's retained records. It is the obligations and nothing else —
// which clock starts when, how far a closure cascades, what a read is allowed to
// be for, what a disposal has to prove — and the records themselves are in a
// [store]. So each obligation is written once and holds wherever the records are
// kept.
type Ledger struct{ store store }
// New returns an empty ledger.
func New() *Ledger {
return &Ledger{
records: make(map[string]*Record),
parties: make(map[string][]string),
inside: make(map[string][]string),
}
}
// New returns a ledger that keeps records in memory, for exercising the
// obligations without a database. What is in it does not survive a restart, so it
// is not what an instance serves from.
func New() *Ledger { return &Ledger{store: newMemory()} }
func partyKey(org, party string) string { return org + "\x00" + party }
// NewBase returns a ledger that keeps records in Base collections, which do
// survive a restart. [Ensure] has to have run first.
func NewBase(app core.App) *Ledger { return &Ledger{store: durable{app: app}} }
// Retain writes a record and returns its id.
//
@@ -54,6 +44,11 @@ func partyKey(org, party string) string { return org + "\x00" + party }
//
// So Record.Start is not an input. A record arriving with one set is refused,
// because a caller that can name its own expiry can name one that never comes.
//
// A retry is not a second fact. A client that resends because it never saw the
// first response gets back the id it already has rather than a second record of
// one transaction — [Record.identify] is what "the same" means for each class,
// and [ErrConflict] is what happens when two different records claim one identity.
func (l *Ledger) Retain(r Record) (string, error) {
if r.Org == "" {
return "", ErrOrg
@@ -62,59 +57,76 @@ func (l *Ledger) Retain(r Record) (string, error) {
return "", fmt.Errorf("%w: the clock is the ledger's to start", ErrClock)
}
l.mu.Lock()
defer l.mu.Unlock()
switch r.Trigger {
case TriggerRefusal, TriggerOccasional:
if r.Relationship != "" {
return "", fmt.Errorf("%w: %s does not run inside a relationship", ErrTrigger, r.Trigger)
}
r.Start = r.Occurred
case TriggerRelationshipEnd:
if r.Relationship == "" {
// The relationship itself. Its clock starts when it ends.
if r.Class != ClassRelationship {
return "", fmt.Errorf("%w: %s needs the relationship it is retained inside", ErrRelationship, r.Class)
var id string
err := l.store.tx(func(s store) error {
switch r.Trigger {
case TriggerRefusal, TriggerOccasional:
if r.Relationship != "" {
return fmt.Errorf("%w: %s does not run inside a relationship", ErrTrigger, r.Trigger)
}
break
r.Start = r.Occurred
case TriggerRelationshipEnd:
if r.Relationship == "" {
// The relationship itself. Its clock starts when it ends.
if r.Class != ClassRelationship {
return fmt.Errorf("%w: %s needs the relationship it is retained inside", ErrRelationship, r.Class)
}
break
}
rel, err := s.read(r.Org, r.Relationship)
if err != nil || rel.Class != ClassRelationship {
return fmt.Errorf("%w: %s", ErrRelationship, r.Relationship)
}
r.Start = rel.Start
default:
return fmt.Errorf("%w: %q", ErrTrigger, r.Trigger)
}
rel, ok := l.records[r.Relationship]
if !ok || rel.Org != r.Org || rel.Class != ClassRelationship {
return "", fmt.Errorf("%w: %s", ErrRelationship, r.Relationship)
}
r.Start = rel.Start
default:
return "", fmt.Errorf("%w: %q", ErrTrigger, r.Trigger)
}
if err := r.validate(); err != nil {
if err := r.validate(); err != nil {
return err
}
// The read and the write are one transaction, so two retries arriving
// together cannot both find nothing and both write.
r.identity = r.identify()
prior, err := s.retained(r.Org, r.identity)
switch {
case err == nil:
if prior.digest() != r.digest() {
return fmt.Errorf("%w: %s", ErrConflict, r.identity)
}
id = prior.ID
return nil
case !errors.Is(err, ErrNotFound):
return err
}
r.ID = uuid.NewString()
r.Written = time.Now().UTC()
// One representation of a moment, everywhere in the ledger.
r.Occurred = r.Occurred.UTC()
if !r.Start.IsZero() {
r.Start = r.Start.UTC()
}
if err := s.insert(r); err != nil {
return err
}
id = r.ID
return nil
})
if err != nil {
return "", err
}
r.ID = uuid.NewString()
r.Written = time.Now().UTC()
// One representation of a moment, everywhere in the ledger.
r.Occurred = r.Occurred.UTC()
if !r.Start.IsZero() {
r.Start = r.Start.UTC()
}
stored := r.clone()
l.records[r.ID] = &stored
for _, p := range r.Parties {
k := partyKey(r.Org, p)
l.parties[k] = append(l.parties[k], r.ID)
}
if r.Relationship != "" {
l.inside[r.Relationship] = append(l.inside[r.Relationship], r.ID)
}
return r.ID, nil
return id, nil
}
// Close ends a business relationship and starts the retention clock on it and on
// every record retained inside it. It returns how many records' clocks started,
// the relationship included: a caller that expects the cascade can check it.
//
// The relationship and the cascade are one write. Half a cascade leaves records
// with no expiry at all, and a record with no expiry is never disposed of — of the
// two ways this can fail, that is the one nobody notices.
func (l *Ledger) Close(org, id string, at time.Time) (int, error) {
if org == "" {
return 0, ErrOrg
@@ -123,36 +135,47 @@ func (l *Ledger) Close(org, id string, at time.Time) (int, error) {
return 0, ErrOccurred
}
l.mu.Lock()
defer l.mu.Unlock()
rel, ok := l.records[id]
if !ok || rel.Org != org || rel.Class != ClassRelationship {
return 0, fmt.Errorf("%w: %s", ErrRelationship, id)
}
if !rel.Ended.IsZero() {
return 0, fmt.Errorf("%w: %s ended %s", ErrClosed, id, rel.Ended.UTC().Format(time.RFC3339))
}
if at.Before(rel.Occurred) {
return 0, fmt.Errorf("%w: %s cannot end before it began", ErrOccurred, id)
}
at = at.UTC()
rel.Ended = at
rel.Start = at
started := 1
for _, child := range l.inside[id] {
r, ok := l.records[child]
if !ok {
continue
started := 0
err := l.store.tx(func(s store) error {
started = 0
rel, err := s.read(org, id)
if err != nil || rel.Class != ClassRelationship {
return fmt.Errorf("%w: %s", ErrRelationship, id)
}
// Only a clock that has not started; an extended or already-started
// record is never moved backwards.
if r.Start.IsZero() {
r.Start = at
if !rel.Ended.IsZero() {
return fmt.Errorf("%w: %s ended %s", ErrClosed, id, rel.Ended.UTC().Format(time.RFC3339))
}
if at.Before(rel.Occurred) {
return fmt.Errorf("%w: %s cannot end before it began", ErrOccurred, id)
}
ended := at.UTC()
rel.Ended, rel.Start = ended, ended
if err := s.update(rel); err != nil {
return err
}
started = 1
retained, err := s.inside(org, id)
if err != nil {
return err
}
for _, r := range retained {
// Only a clock that has not started; an extended or already-started
// record is never moved backwards.
if !r.Start.IsZero() {
continue
}
r.Start = ended
if err := s.update(r); err != nil {
return err
}
started++
}
return nil
})
if err != nil {
return 0, err
}
return started, nil
}
@@ -169,26 +192,25 @@ func (l *Ledger) Extend(org, id string, by time.Duration, reason, who string) er
return fmt.Errorf("%w: needs a positive period, a reason and a decider", ErrExtension)
}
l.mu.Lock()
defer l.mu.Unlock()
return l.store.tx(func(s store) error {
r, err := s.read(org, id)
if err != nil {
return fmt.Errorf("%w: %s", ErrNotFound, id)
}
if r.Start.IsZero() {
return fmt.Errorf("%w: %s", ErrClock, id)
}
if r.Extended != nil {
return fmt.Errorf("%w: %s is already extended", ErrExtension, id)
}
limit := r.Start.AddDate(Period+Extension, 0, 0)
if r.Start.AddDate(Period, 0, 0).Add(by).After(limit) {
return fmt.Errorf("%w: %s past %s", ErrExtension, by, limit.UTC().Format(time.RFC3339))
}
r, ok := l.records[id]
if !ok || r.Org != org {
return fmt.Errorf("%w: %s", ErrNotFound, id)
}
if r.Start.IsZero() {
return fmt.Errorf("%w: %s", ErrClock, id)
}
if r.Extended != nil {
return fmt.Errorf("%w: %s is already extended", ErrExtension, id)
}
limit := r.Start.AddDate(Period+Extension, 0, 0)
if r.Start.AddDate(Period, 0, 0).Add(by).After(limit) {
return fmt.Errorf("%w: %s past %s", ErrExtension, by, limit.UTC().Format(time.RFC3339))
}
r.Extended = &Extended{By: by, Reason: reason, Who: who, At: time.Now().UTC()}
return nil
r.Extended = &Extended{By: by, Reason: reason, Who: who, At: time.Now().UTC()}
return s.update(r)
})
}
// Get returns one record for a permitted purpose. The returned record is a copy:
@@ -200,71 +222,20 @@ func (l *Ledger) Get(p Purpose, org, id string) (Record, error) {
if org == "" {
return Record{}, ErrOrg
}
l.mu.RLock()
defer l.mu.RUnlock()
r, ok := l.records[id]
if !ok || r.Org != org {
return Record{}, fmt.Errorf("%w: %s", ErrNotFound, id)
}
return r.clone(), nil
return l.store.read(org, id)
}
// Each visits an org's records of one class, oldest event first, for a permitted
// purpose. An empty class visits every class. Iteration is ordered so that a
// file produced from it is reproducible.
func (l *Ledger) Each(p Purpose, org string, c Class, fn func(Record) error) error {
func (l *Ledger) Each(p Purpose, org string, c Class, visit func(Record) error) error {
if !p.permitted() {
return fmt.Errorf("%w: %q", ErrPurpose, p)
}
if org == "" {
return ErrOrg
}
// The order is settled first, on the cheap fields, and each record is copied
// as it is visited. A walk therefore does not double the ledger in memory,
// and it does not hold the lock across the caller's callback.
type key struct {
id string
at time.Time
}
l.mu.RLock()
keys := make([]key, 0, len(l.records))
for id, r := range l.records {
if r.Org != org || (c != "" && r.Class != c) {
continue
}
keys = append(keys, key{id: id, at: r.Occurred})
}
l.mu.RUnlock()
slices.SortFunc(keys, func(a, b key) int {
if d := a.at.Compare(b.at); d != 0 {
return d
}
return strings.Compare(a.id, b.id)
})
for _, k := range keys {
l.mu.RLock()
stored, ok := l.records[k.id]
var r Record
if ok {
r = stored.clone()
}
l.mu.RUnlock()
if !ok {
// Disposed between settling the order and reading it. A destroyed
// record is not visited.
continue
}
if err := fn(r); err != nil {
return err
}
}
return nil
return l.store.each(org, c, visit)
}
// Answer answers AMLR Art. 78: whether a business relationship with a named
@@ -313,16 +284,15 @@ func (l *Ledger) Lookback(p Purpose, org, party string, now time.Time) (Answer,
To: now,
}
l.mu.RLock()
defer l.mu.RUnlock()
ids := l.parties[partyKey(org, party)]
answer.Examined = len(ids)
found, err := l.store.party(org, party)
if err != nil {
return Answer{}, err
}
answer.Examined = len(found)
natures := make(map[string]bool)
for _, id := range ids {
r, ok := l.records[id]
if !ok || r.Class != ClassRelationship {
for _, r := range found {
if r.Class != ClassRelationship {
continue
}
// Open now, or ended inside the window.
@@ -335,7 +305,7 @@ func (l *Ledger) Lookback(p Purpose, org, party string, now time.Time) (Answer,
}
answer.Maintained = true
answer.Current = answer.Current || current
answer.Records = append(answer.Records, id)
answer.Records = append(answer.Records, r.ID)
if r.Nature != "" {
natures[r.Nature] = true
}
@@ -370,111 +340,93 @@ type Disposal struct {
// post-conditions and returns an error rather than a count if any of them fails.
// And it deletes whole records only — there is no partial disposal, which is
// what keeps deletion-on-expiry from becoming redaction.
//
// A run works in batches, each of them one write, because the size of a run is
// the ledger's and not one party's. A batch that comes back a second time is a
// batch that survived its own destruction, which is reported rather than retried
// forever.
func (l *Ledger) Dispose(now time.Time) (Disposal, error) {
if err := checkNow(now); err != nil {
return Disposal{}, err
}
l.mu.Lock()
defer l.mu.Unlock()
d := Disposal{At: now.UTC(), Examined: len(l.records)}
var doomed []string
for id, r := range l.records {
if r.Expired(now) {
doomed = append(doomed, id)
}
held, err := l.store.count()
if err != nil {
return Disposal{}, err
}
slices.Sort(doomed)
d := Disposal{At: now.UTC(), Examined: held}
for _, id := range doomed {
r := l.records[id]
delete(l.records, id)
for _, party := range r.Parties {
k := partyKey(r.Org, party)
if rest := drop(l.parties[k], id); len(rest) == 0 {
delete(l.parties, k)
} else {
l.parties[k] = rest
}
var disposed []Record
previous := ""
for {
doomed, err := l.store.expired(now, batch)
if err != nil {
return Disposal{}, err
}
if r.Relationship != "" {
if rest := drop(l.inside[r.Relationship], id); len(rest) == 0 {
delete(l.inside, r.Relationship)
} else {
l.inside[r.Relationship] = rest
}
if len(doomed) == 0 {
break
}
// A disposed relationship stops indexing what was retained inside it.
// Anything of its own that outlives it did so by extension, on purpose.
delete(l.inside, id)
if doomed[0].ID == previous {
return Disposal{}, fmt.Errorf("%w: %s survived its own destruction", ErrDisposal, previous)
}
previous = doomed[0].ID
if err := l.store.tx(func(s store) error { return s.erase(doomed) }); err != nil {
return Disposal{}, err
}
disposed = append(disposed, doomed...)
}
d.Disposed = doomed
if err := l.proveLocked(doomed, now); err != nil {
for _, r := range disposed {
d.Disposed = append(d.Disposed, r.ID)
}
slices.Sort(d.Disposed)
if err := l.prove(disposed, now); err != nil {
return Disposal{}, err
}
return d, nil
}
// proveLocked is the post-condition of a disposal run. It is the difference
// between "the delete statement ran" and "the right records are gone".
func (l *Ledger) proveLocked(disposed []string, now time.Time) error {
gone := make(map[string]bool, len(disposed))
for _, id := range disposed {
gone[id] = true
// prove is the post-condition of a disposal run. It is the difference between "the
// delete statement ran" and "the right records are gone", and it asks the store
// rather than the run's own bookkeeping, because a run that believes its own count
// is exactly the run that hides this bug.
func (l *Ledger) prove(disposed []Record, now time.Time) error {
for _, r := range disposed {
if _, err := l.store.read(r.Org, r.ID); !errors.Is(err, ErrNotFound) {
return fmt.Errorf("%w: %s survived disposal", ErrDisposal, r.ID)
}
retained, err := l.store.inside(r.Org, r.ID)
if err != nil {
return err
}
if len(retained) > 0 {
return fmt.Errorf("%w: %s still indexes %d records retained inside it", ErrDisposal, r.ID, len(retained))
}
for _, p := range r.Parties {
found, err := l.store.party(r.Org, p)
if err != nil {
return err
}
if slices.ContainsFunc(found, func(other Record) bool { return other.ID == r.ID }) {
return fmt.Errorf("%w: %s is still indexed under a party", ErrDisposal, r.ID)
}
}
}
for _, id := range disposed {
if _, ok := l.records[id]; ok {
return fmt.Errorf("%w: %s survived disposal", ErrDisposal, id)
}
if _, ok := l.inside[id]; ok {
return fmt.Errorf("%w: %s still indexes records retained inside it", ErrDisposal, id)
}
// Nothing expired is left anywhere. This is what catches a record the run
// never reached, including one reachable only through an index entry that
// should have gone with something else.
left, err := l.store.expired(now, 1)
if err != nil {
return err
}
for _, ids := range l.parties {
for _, id := range ids {
if gone[id] {
return fmt.Errorf("%w: %s is still indexed under a party", ErrDisposal, id)
}
if _, ok := l.records[id]; !ok {
return fmt.Errorf("%w: party index references vanished record %s", ErrDisposal, id)
}
}
}
for rel, ids := range l.inside {
if _, ok := l.records[rel]; !ok {
return fmt.Errorf("%w: relationship index references vanished record %s", ErrDisposal, rel)
}
for _, id := range ids {
if _, ok := l.records[id]; !ok {
return fmt.Errorf("%w: relationship %s references vanished record %s", ErrDisposal, rel, id)
}
}
}
for id, r := range l.records {
if r.Expired(now) {
return fmt.Errorf("%w: %s is expired and was not disposed", ErrDisposal, id)
}
if len(left) > 0 {
return fmt.Errorf("%w: %s is expired and was not disposed", ErrDisposal, left[0].ID)
}
return nil
}
func drop(ids []string, id string) []string {
out := ids[:0:0]
for _, existing := range ids {
if existing != id {
out = append(out, existing)
}
}
return out
}
// Len is how many records the ledger holds.
func (l *Ledger) Len() int {
l.mu.RLock()
defer l.mu.RUnlock()
return len(l.records)
}
// Len is how many records the ledger holds, in every org.
func (l *Ledger) Len() (int, error) { return l.store.count() }
pkg/retention/memory.go
+270
View File
@@ -0,0 +1,270 @@
// Copyright 2024-2026 Lux Partners Limited. All rights reserved.
// Use of this source code is governed by the Apache 2.0
// license that can be found in the LICENSE file.
package retention
import (
"fmt"
"slices"
"strings"
"sync"
"time"
)
// memory is an in-process store. It exists so the obligations can be exercised at
// the level a reviewer cares about — retain this, close that, does the clock
// cascade — without a database, and it applies exactly the rules the durable store
// applies, because a test against a store with weaker rules than production proves
// nothing about production.
//
// It does not survive a restart, which is why it is not what an instance serves
// from. A retention ledger that does not survive a restart breaches the five-year
// obligation it exists to discharge.
type memory struct {
// mu is nil inside a transaction, where the caller already holds it. Go has no
// re-entrant lock, and the alternative — a second set of methods that do not
// lock — is two ways to do one thing.
mu *sync.Mutex
// shelf is named rather than embedded because the store answers questions
// called inside and party, and so do two of the indexes.
shelf *shelf
}
// shelf is the state a memory store keeps, held by pointer so that the store
// handed to a transaction is the same records under a lock already taken.
type shelf struct {
records map[string]*Record
// party maps org and party to the records naming that party, which is what
// makes the Art. 78 lookback an index read rather than a scan.
party map[string][]string
// inside maps a relationship to the records retained inside it, so ending the
// relationship reaches them.
inside map[string][]string
// identity maps an identity to the record already retained under it, so a
// client's retry finds that record instead of writing a second one.
identity map[string]string
}
func newMemory() *memory {
return &memory{
mu: new(sync.Mutex),
shelf: &shelf{
records: make(map[string]*Record),
party: make(map[string][]string),
inside: make(map[string][]string),
identity: make(map[string]string),
},
}
}
func (m *memory) lock() {
if m.mu != nil {
m.mu.Lock()
}
}
func (m *memory) unlock() {
if m.mu != nil {
m.mu.Unlock()
}
}
// tx runs fn with the lock held, against a store that shares these records. There
// is nothing to roll back: the ledger reports a failed sequence as failed and the
// next run repeats it, which is the same contract the durable store keeps.
func (m *memory) tx(fn func(store) error) error {
m.lock()
defer m.unlock()
return fn(&memory{shelf: m.shelf})
}
func key(org, value string) string { return org + "\x00" + value }
func (m *memory) read(org, id string) (Record, error) {
m.lock()
defer m.unlock()
return m.get(org, id)
}
func (m *memory) get(org, id string) (Record, error) {
r, ok := m.shelf.records[id]
if !ok || r.Org != org {
return Record{}, fmt.Errorf("%w: %s", ErrNotFound, id)
}
return r.clone(), nil
}
func (m *memory) retained(org, identity string) (Record, error) {
m.lock()
defer m.unlock()
id, ok := m.shelf.identity[key(org, identity)]
if !ok {
return Record{}, fmt.Errorf("%w: identity %s", ErrNotFound, identity)
}
return m.get(org, id)
}
func (m *memory) insert(r Record) error {
m.lock()
defer m.unlock()
if _, taken := m.shelf.identity[key(r.Org, r.identity)]; taken {
return fmt.Errorf("%w: %s", ErrConflict, r.identity)
}
stored := r.clone()
m.shelf.records[r.ID] = &stored
m.shelf.identity[key(r.Org, r.identity)] = r.ID
for _, p := range r.Parties {
k := key(r.Org, p)
m.shelf.party[k] = append(m.shelf.party[k], r.ID)
}
if r.Relationship != "" {
k := key(r.Org, r.Relationship)
m.shelf.inside[k] = append(m.shelf.inside[k], r.ID)
}
return nil
}
func (m *memory) update(r Record) error {
m.lock()
defer m.unlock()
stored, ok := m.shelf.records[r.ID]
if !ok || stored.Org != r.Org {
return fmt.Errorf("%w: %s", ErrNotFound, r.ID)
}
kept := r.clone()
m.shelf.records[r.ID] = &kept
return nil
}
func (m *memory) erase(rs []Record) error {
m.lock()
defer m.unlock()
for _, r := range rs {
delete(m.shelf.records, r.ID)
delete(m.shelf.identity, key(r.Org, r.identity))
for _, p := range r.Parties {
k := key(r.Org, p)
if rest := drop(m.shelf.party[k], r.ID); len(rest) == 0 {
delete(m.shelf.party, k)
} else {
m.shelf.party[k] = rest
}
}
if r.Relationship != "" {
k := key(r.Org, r.Relationship)
if rest := drop(m.shelf.inside[k], r.ID); len(rest) == 0 {
delete(m.shelf.inside, k)
} else {
m.shelf.inside[k] = rest
}
}
// A destroyed relationship stops indexing what was retained inside it.
// Anything of its own that outlives it did so by extension, on purpose.
delete(m.shelf.inside, key(r.Org, r.ID))
}
return nil
}
func (m *memory) inside(org, relationship string) ([]Record, error) {
m.lock()
defer m.unlock()
return m.collect(org, m.shelf.inside[key(org, relationship)])
}
func (m *memory) party(org, party string) ([]Record, error) {
m.lock()
defer m.unlock()
return m.collect(org, m.shelf.party[key(org, party)])
}
func (m *memory) collect(org string, ids []string) ([]Record, error) {
out := make([]Record, 0, len(ids))
for _, id := range ids {
r, err := m.get(org, id)
if err != nil {
// An index entry pointing at a record that is not there. It is not this
// read's business to repair it; disposal is what proves the indexes,
// and it reports exactly this as a failure.
continue
}
out = append(out, r)
}
return out, nil
}
// each visits an org's records of one class, oldest event first. The order is
// settled first on the cheap fields and each record is copied as it is visited, so
// a walk does not double the ledger in memory and does not hold the lock across
// the caller's callback.
func (m *memory) each(org string, c Class, visit func(Record) error) error {
type place struct {
id string
at time.Time
}
m.lock()
order := make([]place, 0, len(m.shelf.records))
for id, r := range m.shelf.records {
if r.Org != org || (c != "" && r.Class != c) {
continue
}
order = append(order, place{id: id, at: r.Occurred})
}
m.unlock()
slices.SortFunc(order, func(a, b place) int {
if d := a.at.Compare(b.at); d != 0 {
return d
}
return strings.Compare(a.id, b.id)
})
for _, p := range order {
m.lock()
r, err := m.get(org, p.id)
m.unlock()
if err != nil {
// Destroyed between settling the order and reading it. A record that no
// longer exists is not visited.
continue
}
if err := visit(r); err != nil {
return err
}
}
return nil
}
func (m *memory) expired(now time.Time, limit int) ([]Record, error) {
m.lock()
defer m.unlock()
var doomed []Record
for _, r := range m.shelf.records {
if r.Expired(now) {
doomed = append(doomed, r.clone())
}
}
slices.SortFunc(doomed, func(a, b Record) int { return strings.Compare(a.ID, b.ID) })
if limit > 0 && len(doomed) > limit {
doomed = doomed[:limit]
}
return doomed, nil
}
func (m *memory) count() (int, error) {
m.lock()
defer m.unlock()
return len(m.shelf.records), nil
}
func drop(ids []string, id string) []string {
out := ids[:0:0]
for _, existing := range ids {
if existing != id {
out = append(out, existing)
}
}
return out
}
pkg/retention/record.go
+92
View File
@@ -50,9 +50,15 @@
package retention
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"slices"
"time"
"github.com/google/uuid"
)
// The retention arithmetic, in years, each from its own instrument.
@@ -92,6 +98,8 @@ var (
ErrFuture = errors.New("retention: refusing to act on a future date")
ErrExtension = errors.New("retention: extension exceeds the five further years allowed")
ErrDisposal = errors.New("retention: disposal could not be proven")
ErrConflict = errors.New("retention: a different record is already retained under this identity")
ErrStore = errors.New("retention: the ledger could not be read or written")
)
// Class is what a record is.
@@ -270,6 +278,90 @@ type Record struct {
Assessment *Assessment `json:"assessment,omitempty"`
Written time.Time `json:"written"`
// identity is what makes this record the same retained fact as another. It is
// the ledger's to compute, like the clock, so it is not something a caller can
// name. See [Record.identify].
identity string
}
// identify is the record's identity: the thing that, repeated, means a client
// retried rather than that something happened twice.
//
// What is unique is a property of the class, and it is not unconditional:
//
// - A transaction, a refusal and a relationship are identified by the reference
// they retain. One record per transaction, per refused transaction, per
// relationship — a resubmission of the same reference finds the record it
// already wrote instead of a second copy of it.
// - An assessment recurs. The same case is assessed again whenever new
// information arrives, and every one of those decisions is retained whether or
// not it produced a report (AMLR Art. 77(1)(b)), so the identity includes who
// decided and when. Two decisions on one case are two records; a retry of one
// decision is one.
// - A record with no reference cannot be recognised, because there is nothing to
// recognise it by. It is unique per write, which means a retry of it does
// duplicate. Every writer in this codebase supplies a reference; validate
// requires one of a transaction outright.
func (r Record) identify() string {
if r.Ref == "" {
return uuid.NewString()
}
if r.Class == ClassAssessment && r.Assessment != nil {
return fmt.Sprintf("%s:%s:%s:%d", r.Class, r.Ref, r.Assessment.By, r.Assessment.At.UTC().UnixMilli())
}
return string(r.Class) + ":" + r.Ref
}
// digest is the retained fact, hashed. Two records with one identity are a retry
// only if they say the same thing; if they do not, the caller has two different
// facts under one name, and the ledger refuses rather than choosing between them
// or rewriting what it holds.
//
// Only what the caller supplied is hashed. The id, the clock and the write time
// are the ledger's, and hashing them would make every retry a conflict. Times are
// taken to the millisecond they are stored at, so a record and the same record
// read back digest alike.
func (r Record) digest() string {
type fact struct {
Class Class
Trigger Trigger
Parties []string
Ref string
Nature string
Reason string
Inside string
Occurred int64
Body []byte
Assessment *Assessment
}
f := fact{
Class: r.Class,
Trigger: r.Trigger,
Parties: slices.Sorted(slices.Values(r.Parties)),
Ref: r.Ref,
Nature: r.Nature,
Reason: r.Reason,
Inside: r.Relationship,
Occurred: r.Occurred.UTC().UnixMilli(),
Body: r.Body,
Assessment: r.Assessment,
}
if f.Assessment != nil {
a := f.Assessment.clone()
a.At = a.At.UTC().Truncate(time.Millisecond)
f.Assessment = &a
}
// A struct marshals in field order, so the encoding is stable; a map would not
// be. Marshalling cannot fail on these types, and a digest that silently became
// the empty string would make every record a retry of every other, so the
// error is not ignored.
encoded, err := json.Marshal(f)
if err != nil {
return fmt.Sprintf("unhashable:%v", err)
}
sum := sha256.Sum256(encoded)
return hex.EncodeToString(sum[:])
}
// Expiry is when the record must be destroyed. Zero means the retention clock
pkg/retention/schema.go
+110
View File
@@ -0,0 +1,110 @@
// Copyright 2024-2026 Lux Partners Limited. All rights reserved.
// Use of this source code is governed by the Apache 2.0
// license that can be found in the LICENSE file.
package retention
import (
"github.com/hanzoai/base/core"
"github.com/luxfi/aml/pkg/store"
)
// Field names. They are named once here because the writer and the reader have to
// agree: a disagreement between them does not fail, it returns nothing, and a
// ledger that returns nothing looks exactly like a customer with no history.
const (
fieldClass = "class"
fieldTrigger = "trigger"
fieldParties = "parties"
fieldRef = "ref"
fieldNature = "nature"
fieldReason = "reason"
fieldInside = "relationship"
fieldOccurred = "occurred"
fieldEnded = "ended"
fieldStart = "start"
fieldExpiry = "expiry"
fieldWritten = "written"
fieldBody = "body"
fieldAssess = "assessment"
fieldExtended = "extended"
fieldIdentity = "identity"
// On the party index only: which party, and which record names it.
fieldParty = "party"
fieldRecord = "record"
)
// records is where retained records live.
//
// Expiry is stored although [Record.Expiry] computes it, because disposal has to
// find what has expired without reading every record to ask. It is derived in one
// place — the writer, from that one function — so the column cannot say something
// the arithmetic does not.
var records = store.Kind{
Name: "aml_retention",
Fields: []core.Field{
&core.TextField{Name: fieldClass, Required: true},
&core.TextField{Name: fieldTrigger, Required: true},
&core.JSONField{Name: fieldParties, Required: true},
&core.TextField{Name: fieldRef},
&core.TextField{Name: fieldNature},
&core.TextField{Name: fieldReason},
&core.TextField{Name: fieldInside},
&core.DateField{Name: fieldOccurred, Required: true},
&core.DateField{Name: fieldEnded},
&core.DateField{Name: fieldStart},
&core.DateField{Name: fieldExpiry},
&core.DateField{Name: fieldWritten, Required: true},
&core.TextField{Name: fieldBody},
&core.JSONField{Name: fieldAssess},
&core.JSONField{Name: fieldExtended},
&core.TextField{Name: fieldIdentity, Required: true},
},
Indexes: []store.Index{
// A client retry must not retain the same fact twice. The identity is
// unique per org and the database is what makes it so, because two
// concurrent retries both read no prior record before either writes.
{Name: "identity", Fields: []string{store.Org, fieldIdentity}, Unique: true},
// Closing a relationship cascades the clock to everything retained inside
// it, so that set has to be reachable without a scan.
{Name: "inside", Fields: []string{store.Org, fieldInside}},
// The ordered walk a file is produced from, oldest event first.
{Name: "class", Fields: []string{store.Org, fieldClass, fieldOccurred}},
// Disposal asks which records have expired, in every org.
{Name: "expiry", Fields: []string{fieldExpiry}},
},
}
// parties indexes records by the parties they name. It is a collection of its own
// because a party is one of several on a record, and a value that repeats cannot
// be a column that is indexed.
//
// AMLR Art. 78 has to be answered "fully and speedily", which is this index: a
// lookback reads one party's records and its cost does not grow with the ledger.
var parties = store.Kind{
Name: "aml_retention_parties",
Fields: []core.Field{
&core.TextField{Name: fieldParty, Required: true},
&core.TextField{Name: fieldRecord, Required: true},
},
Indexes: []store.Index{
// The lookback: which records name this party.
{Name: "party", Fields: []string{store.Org, fieldParty}, Unique: false},
// Disposal: which index entries reference this record, so they go with it.
// Unique because one record names a party once.
{Name: "record", Fields: []string{store.Org, fieldRecord, fieldParty}, Unique: true},
},
}
// Ensure creates what the ledger stores in, and is idempotent so it runs on every
// start. A retention ledger that does not survive a restart breaches the
// record-keeping obligation it exists to discharge (AMLR Art. 77), so this is the
// call that has to have happened before the first record is retained.
func Ensure(app core.App) error {
if err := records.Ensure(app); err != nil {
return err
}
return parties.Ensure(app)
}
pkg/retention/store.go
+51
View File
@@ -0,0 +1,51 @@
// Copyright 2024-2026 Lux Partners Limited. All rights reserved.
// Use of this source code is governed by the Apache 2.0
// license that can be found in the LICENSE file.
package retention
import "time"
// store is where a ledger's records are kept.
//
// The [Ledger] holds the obligations — which clock starts when, how far a closure
// cascades, what a disposal has to prove before it reports success — and the store
// holds the records. So every obligation is stated once and holds whichever store
// is underneath, and a store cannot weaken one by implementing it differently.
//
// The contract:
//
// - Every method is scoped to an org, except the two that answer to the operator
// rather than to a tenant: expired and count. Destroying what has expired is
// the operator's obligation wherever the record sits, and there is no list of
// orgs to iterate.
// - Methods promise their own single effect and nothing more. A sequence that
// must be all-or-nothing — read a relationship then write its children — is
// wrapped in tx by the ledger, and inside tx the store is one that commits
// everything or nothing.
// - read and retained report [ErrNotFound] for an absent record, so a caller can
// tell "not there" from "could not look".
// - Reads return whole records, copied. There is no operation that returns part
// of one, because the unit of disposal is the whole record and a store that
// could hand out a piece could hand out a record with a piece missing.
type store interface {
tx(fn func(store) error) error
read(org, id string) (Record, error)
retained(org, identity string) (Record, error)
insert(r Record) error
update(r Record) error
erase(rs []Record) error
inside(org, relationship string) ([]Record, error)
party(org, party string) ([]Record, error)
each(org string, c Class, visit func(Record) error) error
expired(now time.Time, limit int) ([]Record, error)
count() (int, error)
}
// batch is how many records a disposal run destroys at a time. Disposal is the one
// operation whose size is the ledger's rather than one party's, so it is done in
// bounded pieces instead of assembling five years of records in memory.
const batch = 500
pkg/store/store.go
+189
View File
@@ -0,0 +1,189 @@
// Copyright 2024-2026 Lux Partners Limited. All rights reserved.
// Use of this source code is governed by the Apache 2.0
// license that can be found in the LICENSE file.
// Package store is how this codebase keeps anything that has to survive a
// restart. There is one way, and this is it.
//
// A durable store is a [Kind]: a collection's name, the fields it holds, and the
// indexes that follow the queries it will be asked. [Kind.Ensure] creates it and
// brings an existing one up to the declaration, so it runs on every start and the
// schema the reader queries is the schema that exists. Written as a migration
// instead, the two drift, and a drifted schema does not fail — it returns nothing,
// which a monitoring system reads as nothing to report.
//
// Three rules, each held up by the code rather than by discipline:
//
// - Every collection carries [Org]. Ensure adds it, so a Kind cannot be
// declared without a tenant, and [Kind.New] sets it, so a record cannot be
// written without one.
// - Every read is scoped. [Kind.Find] ands the tenant predicate onto the
// caller's filter and refuses an empty org. A query that has to cross tenants
// says [Kind.Across] instead, so the whole set of them is one grep.
// - Field names are constants in the package that declares the Kind, so the
// writer and the reader name the same column and a rename cannot compile on
// one side only. A silent disagreement between them produces an empty result,
// and an empty result is indistinguishable from an empty ledger.
//
// Values are bound as parameters, never interpolated. A value that reaches the
// filter as text can change the shape of the query: it can widen the result to
// another tenant's records or narrow it to none, and narrowing it to none is the
// dangerous direction.
package store
import (
"fmt"
"strings"
"github.com/hanzoai/base/core"
"github.com/hanzoai/dbx"
)
// Org is the field that carries the tenant. Every Kind has it.
const Org = "org_id"
// orgParam is what the tenant predicate binds to. It is deliberately not "org",
// so that a caller's own parameter can never shadow it.
const orgParam = "store_org"
// Index is one index over a Kind, named for the query it serves rather than for
// the fields it happens to contain.
type Index struct {
Name string
Fields []string
Unique bool
}
// Kind declares one collection: what it is called, what it holds, and the
// indexes its queries need. It is a value, and the app is passed to each
// operation, so the same Kind serves a plain app and a transaction.
type Kind struct {
Name string
Fields []core.Field
Indexes []Index
}
// Ensure creates the collection if it is absent and adds anything the
// declaration has that the existing collection lacks. It is idempotent, so it
// runs on every start, and it is additive: it never drops a field or an index,
// because a field this binary does not know about may be the one holding a record
// another obligation depends on.
func (k Kind) Ensure(app core.App) error {
collection, err := app.FindCollectionByNameOrId(k.Name)
if err != nil {
collection = core.NewBaseCollection(k.Name)
collection.Fields.Add(&core.TextField{Name: Org, Required: true})
}
changed := false
for _, f := range k.Fields {
if collection.Fields.GetByName(f.GetName()) == nil {
collection.Fields.Add(f)
changed = true
}
}
for _, i := range k.Indexes {
if !indexed(collection.Indexes, k.name(i)) {
collection.Indexes = append(collection.Indexes, k.statement(i))
changed = true
}
}
if !changed && !collection.IsNew() {
return nil
}
if err := app.Save(collection); err != nil {
return fmt.Errorf("store: %s: %w", k.Name, err)
}
return nil
}
// New builds an unsaved record already scoped to an org, which is the only way to
// get one: a record whose tenant is decided after the fact is a record that can
// be written to the wrong one.
func (k Kind) New(app core.App, org string) (*core.Record, error) {
if org == "" {
return nil, fmt.Errorf("store: %s: refusing to write a record that belongs to no organisation", k.Name)
}
collection, err := app.FindCollectionByNameOrId(k.Name)
if err != nil {
return nil, fmt.Errorf("store: %s: %w", k.Name, err)
}
r := core.NewRecord(collection)
r.Set(Org, org)
return r, nil
}
// Find returns one org's records. The tenant predicate is added here rather than
// left to the caller's filter, so a forgotten clause cannot read across tenants.
// A limit of 0 returns everything matched.
//
// There is no offset. A walk that pages by offset over a collection something
// else is deleting from skips rows, and skipping a row in a retention ledger is
// how a record outlives its period unnoticed; page by the last key seen instead.
func (k Kind) Find(app core.App, org, filter, sort string, limit int, params dbx.Params) ([]*core.Record, error) {
if org == "" {
return nil, fmt.Errorf("store: %s: refusing an unscoped query", k.Name)
}
if params == nil {
params = dbx.Params{}
}
params[orgParam] = org
scoped := Org + " = {:" + orgParam + "}"
if filter != "" {
scoped += " && (" + filter + ")"
}
return k.records(app, scoped, sort, limit, params)
}
// Across reads every org's records. It exists for the operations an obligation
// puts on the operator rather than on a tenant — destroying what has expired
// wherever it is — and it is named so that every crossing of the tenant boundary
// is found by searching for one word.
func (k Kind) Across(app core.App, filter, sort string, limit int, params dbx.Params) ([]*core.Record, error) {
return k.records(app, filter, sort, limit, params)
}
// Count is how many records the collection holds, in every org.
func (k Kind) Count(app core.App) (int, error) {
n, err := app.CountRecords(k.Name)
if err != nil {
return 0, fmt.Errorf("store: %s: %w", k.Name, err)
}
return int(n), nil
}
func (k Kind) records(app core.App, filter, sort string, limit int, params dbx.Params) ([]*core.Record, error) {
out, err := app.FindRecordsByFilter(k.Name, filter, sort, limit, 0, params)
if err != nil {
return nil, fmt.Errorf("store: %s: %w", k.Name, err)
}
return out, nil
}
// name is an index's full name, which is unique across collections because an
// index name is a schema-wide identifier.
func (k Kind) name(i Index) string { return "idx_" + k.Name + "_" + i.Name }
func (k Kind) statement(i Index) string {
unique := ""
if i.Unique {
unique = "UNIQUE "
}
cols := make([]string, 0, len(i.Fields))
for _, f := range i.Fields {
cols = append(cols, "`"+f+"`")
}
return fmt.Sprintf("CREATE %sINDEX `%s` ON `%s` (%s)", unique, k.name(i), k.Name, strings.Join(cols, ", "))
}
// indexed reports whether an index of that name already exists, by name rather
// than by statement: the stored statement is normalised on save, so comparing
// text would declare every index missing and try to create it again.
func indexed(existing []string, name string) bool {
for _, s := range existing {
if strings.Contains(s, "`"+name+"`") {
return true
}
}
return false
}