world

package
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 13, 2026 License: Apache-2.0, MIT Imports: 31 Imported by: 0

Documentation

Overview

Package world owns the simulator's on-disk state: the pebble db, the global RNG, the deterministic account roster, and the live commit generator. The HTTP layer reads through *World; only the traffic goroutine writes to pebble after bootstrap.

Index

Constants

View Source 
const MalformedIdentityDID = "did:plc:oracle!malformed"

MalformedIdentityDID is the syntactically-invalid DID carried by GenerateMalformedIdentityForTest frames. Exported so oracle asserts can locate the archived row.

Variables

View Source 
var ErrDataDirReserved = errors.New("world: --data-dir cannot be ./data; use ./data/simulator")

ErrDataDirReserved is returned by New when DataDir resolves to the jetstream data directory. The simulator owns its own pebble db and must never share a directory with the production binary.

View Source 
var ErrSeedMismatch = errors.New("world: seed mismatch; pass --reset or restore previous --seed")

ErrSeedMismatch is returned by EnsureSeed when the persisted seed does not match cfg.Seed. Operators must --reset (or change cfg.Seed back) before continuing.

Functions

func EncodeOutdatedCursorInfo 

func EncodeOutdatedCursorInfo() []byte

EncodeOutdatedCursorInfo returns a wire-format #info frame signalling OutdatedCursor. The relay handler sends this before falling back to live streaming when a consumer's cursor is older than the retained history.

func VirtualPDSHostname 

func VirtualPDSHostname(index int) string

VirtualPDSHostname is the stable hostname used by the in-process simulator.

Types

type Account 

type Account struct {
	Index int
	DID   atmos.DID
	// contains filtered or unexported fields
}

Account is the exported view of a simulator account, for HTTP handlers and tests living outside this package. Internal code (everything else in package world) uses the unexported `account` directly.

func (Account) HandleSuffix 

func (a Account) HandleSuffix() string

HandleSuffix is the cosmetic handle disambiguator: just the index.

func (Account) PubkeyMultibase 

func (a Account) PubkeyMultibase() string

PubkeyMultibase returns the z-prefixed base58 multibase encoding of the account's atproto signing key.

type AdversarialEntry 

type AdversarialEntry struct {
	Source     AdversarialSource
	Layer      AdversarialLayer
	Reason     string
	Seq        int64 // firehose seq of the lying frame; 0 for backfill-only lies
	DID        string
	Collection string
	Rkey       string
	WholeEvent bool
}

AdversarialEntry is one recorded lie. Reason carries the expected drop-reason label for gate-owned lies (matching jetstream's ingest.DropReason values: "invalid_rev", "invalid_collection", "invalid_rkey", "field_too_long") and a descriptive tag for verifier-owned ones. WholeEvent marks lies that drop the entire event (every row of the seq) rather than a single op.

type AdversarialLayer 

type AdversarialLayer string

AdversarialLayer labels which layer of the consuming stack is expected to reject the lie. Gate-owned lies land on jetstream's shared drop counter with a specific reason; verifier-owned lies are rejected or repaired by atmos's Sync-1.1 verifier before the gate. 

const (
	AdversarialLayerGate     AdversarialLayer = "gate"
	AdversarialLayerVerifier AdversarialLayer = "verifier"
)

type AdversarialLedger 

type AdversarialLedger struct {
	// contains filtered or unexported fields
}

AdversarialLedger accumulates every lie the world told, in emission order, plus a key index so honest traffic can refuse to touch lie records (see pickUntouchedRecord).

func (*AdversarialLedger) ContainsKey 

func (l *AdversarialLedger) ContainsKey(key string) bool

ContainsKey reports whether any recorded lie carries the MST key (collection/rkey form). Honest traffic generators consult this so they never mutate a lie record: a spec-valid-but-unrepresentable key (e.g. a 300-byte rkey) passes every spec check, but an honest single-op commit touching it would be gate-dropped whole and its cursor would never be archived — starving the oracle's gap-free cursor accounting on an event the ledger never promised to drop.

func (*AdversarialLedger) Entries 

func (l *AdversarialLedger) Entries() []AdversarialEntry

Entries returns a copy of all recorded lies in emission order.

type AdversarialSource 

type AdversarialSource string

AdversarialSource labels which ingest path a recorded lie targets. 

const (
	AdversarialSourceLive     AdversarialSource = "live"
	AdversarialSourceBackfill AdversarialSource = "backfill"
)

type Config 

type Config struct {
	DataDir  string
	Reset    bool
	Seed     uint64
	Accounts int
	// PDSHosts is the number of virtual PDSes that own Accounts. Zero uses
	// the default four-host skewed topology.
	PDSHosts          int
	InitialRecords    int
	InitialRecordsMin int
	InitialRecordsMax int
	CommitsPerSec     float64
	RateMultiplier    float64
	FirehoseHistory   int
	TrafficMix        TrafficMix
}

Config drives *World construction.

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig returns simulator defaults matching the design doc.

type GeneratedChainOp 

type GeneratedChainOp struct {
	Action     string
	Collection string
	Rkey       string
	Rev        string
	Payload    []byte
}

GeneratedChainOp describes one op injected via GenerateRecordOpForTest, carrying enough detail for a test to derive the durable event-log row it should produce: the action, its (collection, rkey), the rev the commit assigned, and the record's CBOR block (nil for delete). Payload equals the record block jetstream records on disk for create/update.

type ListReposEntry 

type ListReposEntry struct {
	DID    atmos.DID
	Rev    string
	Head   string // commit CID string
	Active bool
}

ListReposEntry is one row of a listRepos response.

type TargetedOpSpec 

type TargetedOpSpec struct {
	Action     string
	Collection string
	Rkey       string
	StripBlock bool
}

TargetedOpSpec describes one op in a GenerateMultiOpCommitForTest commit. StripBlock excludes the op's record leaf block from the broadcast CAR diff — the wire op still references the block's CID, so the frame carries the partial-CAR shape a non-canonical PDS emits (spec-permitted; the record is unarchivable from the frame alone). Only valid on create/update: deletes carry no block.

type TrafficMix 

type TrafficMix struct {
	Create   float64
	Update   float64
	Delete   float64
	Identity float64
}

TrafficMix is the weighted event-kind distribution the live traffic pump draws from. Weights are relative, not percentages. It is a Config field (rather than a package constant) so future swarm-style tiers can draw a different mix per seed (#233).

The commit-action weights are deliberately NOT production-shaped: a 180s production sample (2026-07-04) measured create 95.5 / delete 3.9 / update 0.6 and identity at 0.061% of all events. The mix over-weights tombstone-forming ops (update/delete) because that is where compaction bugs live, and holds identity well above its production rate so a default-scale oracle run (~200 live events) still exercises the path several times instead of 0.12 times. Production-shaped regression coverage is the corpus tier's job.

func DefaultTrafficMix 

func DefaultTrafficMix() TrafficMix

DefaultTrafficMix returns the design-doc action distribution plus the identity weight discussed on #202.

type World 

type World struct {
	// contains filtered or unexported fields
}

World is the simulator's runtime handle: pebble db + the in-memory state that derives from it. Goroutine-safety: pebble itself is safe; mutationMu serializes post-bootstrap event generation, including the shared RNG and logical-clock state. Sequence allocation is via atomic.Int64.

func New 

func New(_ context.Context, cfg Config) (*World, error)

New opens (creating if needed) the simulator pebble db at cfg.DataDir. With cfg.Reset = true, removes the directory first. Refuses to operate when cfg.DataDir resolves to "./data".

func (*World) AccountCount 

func (w *World) AccountCount() int

AccountCount returns the total accounts in the world.

func (*World) AccountIndicesForTest 

func (w *World) AccountIndicesForTest() ([]int, error)

AccountIndicesForTest returns every account index persisted in the world, including hidden test accounts that AccountCount/ListReposPage intentionally omit.

func (*World) AddHiddenAccountForTest 

func (w *World) AddHiddenAccountForTest(ctx context.Context, initialRecords int) (int, Account, error)

AddHiddenAccountForTest creates a real simulator account outside the listRepos roster. getRepo/PLC lookup can find it by DID, and it can emit signed live traffic, but AccountCount/ListReposPage still expose only the original cfg.Accounts accounts. This is useful for tests that need a repo reachable by DID but deliberately omitted from listRepos.

func (*World) AdversarialLedger 

func (w *World) AdversarialLedger() *AdversarialLedger

AdversarialLedger exposes the world's lie ledger for oracle reconciliation.

func (*World) AttachRuntime 

func (w *World) AttachRuntime(r *rand.Rand, fan *fanout.Registry) error

AttachRuntime wires in the live RNG and fanout. Called once after New + EnsureSeed + Bootstrap by cmd/simulator's serve action.

func (*World) Bootstrap 

func (w *World) Bootstrap(ctx context.Context, logger *slog.Logger) error

Bootstrap generates and persists per-account initial records. Idempotent: state rows already at the target shape are not rewritten, so re-running on a partially-populated db is safe.

Uses a dedicated PCG seeded from cfg.Seed for the *content* of initial records. The runtime RNG owned by *World drives only live traffic; mixing the two would make resume-from-disk content-dependent on whether a previous run had bootstrapped fully.

func (*World) Close 

func (w *World) Close() error

Close releases the pebble db. Idempotent.

func (*World) CurrentSeq 

func (w *World) CurrentSeq() int64

CurrentSeq returns the latest persisted firehose seq.

func (*World) EnsureSeed 

func (w *World) EnsureSeed() (wantBootstrap bool, err error)

EnsureSeed implements the seed handshake:

  • first run (no row): persists cfg.Seed, returns (true, nil) → "caller should run bootstrap"
  • matching row: returns (false, nil) → "resume"
  • mismatched row: returns (_, ErrSeedMismatch)

func (*World) ExportRepoCAR 

func (w *World) ExportRepoCAR(idx int, dst io.Writer) error

ExportRepoCAR writes the account's persisted repo head as a CAR. Unlike repo.ExportCAR, this does not sign a fresh commit; getRepo must expose the same head CID and rev that listRepos advertised.

func (*World) FindAccountByDID 

func (w *World) FindAccountByDID(did atmos.DID) (Account, bool, error)

FindAccountByDID returns (account, true) if a matching account exists; (Account{}, false, nil) otherwise. Linear scan over the account/<idx>/did rows; acceptable at 10k accounts because the simulator caches identity resolutions through atmos's directory cache anyway.

func (*World) FirehoseRange 

func (w *World) FirehoseRange(cursor int64, limit int) ([][]byte, error)

FirehoseRange exposes the read-side of the ring buffer for relay subscribers (Task 15).

func (*World) GenerateAccountDeleteForTest 

func (w *World) GenerateAccountDeleteForTest(ctx context.Context, idx int) ([]byte, error)

func (*World) GenerateAccountReactivateForTest 

func (w *World) GenerateAccountReactivateForTest(ctx context.Context, idx int) ([]byte, error)

GenerateAccountReactivateForTest clears a deleted account's flag and emits an Active:true #account frame, re-enabling commits. Oracle tests use it for the DID-level no-permanent-tombstone path.

func (*World) GenerateAccountStatusForTest 

func (w *World) GenerateAccountStatusForTest(ctx context.Context, idx int, active bool, status string) ([]byte, error)

GenerateAccountStatusForTest emits a #account frame with the caller-supplied active/status pair without mutating the world's repo or deleted flag. Oracle tests use this to pin non-deleted hosting statuses end-to-end: only Active:false,status:"deleted" is a tombstone.

func (*World) GenerateAdversarialOpForTest 

func (w *World) GenerateAdversarialOpForTest(ctx context.Context, idx int, badKey, reason string) (GeneratedChainOp, error)

GenerateAdversarialOpForTest emits one #commit frame carrying TWO create ops: a benign sibling on a fresh honest path, and a lie whose raw MST key is the caller-supplied badKey (full "collection/rkey" form, NOT validated). The lie is inserted with mst.Tree.Insert — bypassing repo.Create's spec validation — so the signed MST, the CAR diff, and the wire op all agree and the commit verifies cleanly.

The sibling is the survivors-contract probe: the oracle asserts it archives even though the lie in the same commit drops. Returns the sibling's GeneratedChainOp (the row the oracle should find durable).

reason must be the drop-reason label the ingest gate is expected to emit for badKey ("invalid_collection", "invalid_rkey", or "field_too_long" for spec-valid-but-unrepresentable keys).

func (*World) GenerateAdversarialSyncForTest 

func (w *World) GenerateAdversarialSyncForTest(ctx context.Context, idx int, badRev string) ([]byte, error)

GenerateAdversarialSyncForTest silently mutates account idx (no #commit frame), then emits a #sync frame whose ENVELOPE rev is the caller-supplied lie. The silent mutation is load-bearing: it makes the sync's data CID diverge from the consumer's chain state, which is the only route to the gate —

  • rev lexically <= chain state's rev → the verifier's rev-replay check silently drops the frame (empty rev always lands here);
  • rev above state but data MATCHING → the verifier's no-op fast path cross-checks envelope vs inner rev → FieldMismatchError, verifier-owned;
  • rev above state and data DIVERGENT → the verifier resyncs (fetches the authoritative repo) and yields ops; the event — still carrying the lying envelope rev — reaches jetstream's convertSync where validateRev drops the WHOLE event ({live, invalid_rev}).

badRev must be unparseable as a TID and lexically greater than every TID (start it with a byte above 'j', e.g. "not-a-tid") so the replay check cannot eat it.

PERMANENT ARCHIVAL LOSS, by design: the verifier's resync repairs its own chain state to the post-mutation head, so a later honest #sync at the same rev is replay-dropped — the silently-created record's only carrier was the dropped event. The record is ledgered (dropped-op coordinates + whole-event seq) so the oracle excludes it from ground truth and cursor-gap accounting; this is exactly the documented loss semantics of refusing spec-invalid input.

func (*World) GenerateIdentityForTest 

func (w *World) GenerateIdentityForTest(ctx context.Context, idx int, handleChange bool) ([]byte, error)

GenerateIdentityForTest emits one polite #identity frame for account idx: handle-absent (the dominant production shape) or, with handleChange, a handle-change payload backed by the account's persisted change counter. Oracle tests use it to pin deterministic identity coverage independent of the random traffic mix.

func (*World) GenerateMalformedIdentityForTest 

func (w *World) GenerateMalformedIdentityForTest(ctx context.Context) ([]byte, error)

GenerateMalformedIdentityForTest emits an #identity frame whose DID (MalformedIdentityDID) fails atproto DID syntax, modeling the unverified-upstream reality that #identity bodies are not signature-checked by relays. Injection-only adversarial input — the random traffic mix never produces it.

func (*World) GenerateMultiOpCommitForTest 

func (w *World) GenerateMultiOpCommitForTest(ctx context.Context, idx int, specs []TargetedOpSpec) ([]byte, []GeneratedChainOp, error)

GenerateMultiOpCommitForTest applies several targeted ops on account idx in ONE commit, optionally stripping chosen record leaf blocks from the CAR diff (see TargetedOpSpec.StripBlock). Only record leaf blocks are ever stripped — the commit block and every MST node stay in the CAR, so the frame still verifies (atmos's inversion needs the tree, not the leaves) and the fault is precisely "ops whose record block is absent", not a malformed CAR. The world's own persisted repo state includes every op; only the wire frame is partial.

Specs must name distinct (collection, rkey) paths: atmos's verifier rejects duplicate paths in a single commit, and applyTargetedOp's create-on-existing guard would trip anyway for repeated creates.

func (*World) GenerateOneForTest 

func (w *World) GenerateOneForTest(ctx context.Context) ([]byte, error)

GenerateOneForTest exposes generateOne for the http_test package. Production callers use RunTraffic; only tests need to drive individual events synchronously.

func (*World) GenerateRecordOpForTest 

func (w *World) GenerateRecordOpForTest(ctx context.Context, idx int, action, coll, rkey string) ([]byte, GeneratedChainOp, error)

GenerateRecordOpForTest applies a single create/update/delete on account idx against the caller-specified (collection, rkey), commits, and broadcasts the resulting #commit frame on the live firehose. It is the targeted analogue of GenerateOneForTest: where ordinary traffic picks random paths, this lets a test drive an exact chain on a known key — in particular a delete followed by a recreate reusing the SAME rkey, which random traffic (fresh TID rkeys) never produces. Record payloads are still drawn from the world RNG, so payload bytes vary by seed. Returns the wire frame and the op descriptor (assigned rev + record block).

func (*World) GenerateSilentMutationThenCommitForTest 

func (w *World) GenerateSilentMutationThenCommitForTest(ctx context.Context, idx int) ([]byte, error)

GenerateSilentMutationThenCommitForTest mutates account idx, skips that commit frame, then emits the next commit for the same DID. The emitted commit's prevData points at a state Jetstream never saw, forcing the verifier chain-break path and its async resync repair.

func (*World) GenerateSilentMutationThenSyncForTest 

func (w *World) GenerateSilentMutationThenSyncForTest(ctx context.Context, idx int) ([]byte, error)

GenerateSilentMutationThenSyncForTest mutates account idx, intentionally skips publishing the corresponding #commit frame, then emits a #sync for the new repo head. Oracle tests use this to force a true local/upstream divergence: Jetstream must recover the authoritative state via getRepo.

func (*World) GenerateSyncForTest 

func (w *World) GenerateSyncForTest(ctx context.Context, idx int) ([]byte, error)

GenerateSyncForTest emits a real subscribeRepos #sync frame for the current head of account idx. It does not mutate the repo; it packages the current commit block in the #sync CAR body, persists the frame to firehose history, and publishes it to live subscribers.

func (*World) GenerateVerifierRejectedCommitForTest 

func (w *World) GenerateVerifierRejectedCommitForTest(ctx context.Context, idx int, badRev, reason string) ([]byte, error)

GenerateVerifierRejectedCommitForTest emits a #commit frame whose rev is signed-in but invalid at the VERIFIER layer: reason selects the lie shape. These frames never reach the ingest gate — atmos rejects them pre-conversion — so the oracle asserts verifier-failure classification + no archive + cursor advance instead of a gate counter. Supported reasons:

  • "non_tid_rev": rev fails ParseTID (VerifyCommit InvalidRevError)
  • "future_rev": rev is a valid TID > 5m ahead of the consumer's clock (checkFutureRev FutureRevError). The caller supplies the TID via rev since only the test knows the consumer's fake clock.

The commit is otherwise honest: a real create op, real signed MST. The world's persisted head DOES advance to the lying rev, which has two consequences callers must manage:

  1. While the head rev is invalid, a getRepo fetch of this account fails at atmos's repo loader (non-empty invalid rev) or produces gate-dropped rows (empty rev), so a verifier-triggered resync cannot repair the DID yet.
  2. The next HONEST commit on the account restores a valid head; its PrevData points at the lie's MST root, which jetstream never accepted, so the verifier chain-breaks and repairs via resync from the now-honest head. Self-healing, and the repair itself is useful coverage.

Oracle scenarios should therefore follow this call with at least one honest commit on the same account before final-state comparison. The lie's record stays in the world MST (ground truth); the ledger entry lets the oracle exclude it until the follow-up honest commit's resync materializes it. Because the record IS eventually repaired, the entry is recorded with Layer=verifier for cursor-gap exemption only — final-state exclusion must check whether repair happened.

func (*World) InjectAdversarialRecordForBackfill 

func (w *World) InjectAdversarialRecordForBackfill(ctx context.Context, idx int, badKey, reason string) error

InjectAdversarialRecordForBackfill commits a lie into account idx's repo WITHOUT publishing any firehose frame (the silent-mutation precedent). The adversarial key rides the persisted MST, so jetstream's backfill getRepo download walks straight into it and the backfill half of the #197 gate must drop it while archiving the account's honest records. This is also the ONLY route for invalid-UTF-8 rkeys (wire-blocked on the live path; MST node keys are CBOR byte strings and carry arbitrary bytes).

Must be called BEFORE jetstream bootstraps (or before the account's repo is fetched) for the lie to be visible to backfill.

func (*World) IsAccountDeleted 

func (w *World) IsAccountDeleted(idx int) (bool, error)

func (*World) ListReposPage 

func (w *World) ListReposPage(start, limit int) (entries []ListReposEntry, nextStart int, err error)

ListReposPage returns up to limit entries starting at index `start`. nextStart is start + len(entries); when nextStart == AccountCount(), the caller has paged through everything.

func (*World) ListReposPageForPDS 

func (w *World) ListReposPageForPDS(pdsIndex, start, limit int) ([]ListReposEntry, int, error)

ListReposPageForPDS pages one host's authoritative roster. start and the returned nextStart are ordinals in that host's own cursor space.

func (*World) LoadAccount 

func (w *World) LoadAccount(idx int) (Account, error)

LoadAccount returns the account at the given index.

func (*World) LoadRepo 

func (w *World) LoadRepo(idx int) (*repo.Repo, *crypto.K256PrivateKey, error)

LoadRepo returns a fully-loaded *repo.Repo plus the signing key needed to call ExportCAR. Reads MST/record blocks lazily from pebble; safe to call concurrently because the underlying pebbleStore only reads.

func (*World) PDSAccountCount 

func (w *World) PDSAccountCount(pdsIndex int) int

PDSAccountCount returns the authoritative direct-listRepos count for a host.

func (*World) PDSHostCount 

func (w *World) PDSHostCount() int

PDSHostCount returns the number of virtual PDSes in this world.

func (*World) PDSIndexForAccount 

func (w *World) PDSIndexForAccount(accountIdx int) int

PDSIndexForAccount deterministically assigns an account to a virtual PDS. Host zero receives roughly 60% of accounts (the "big mushroom"); the tail is spread uniformly across the remaining hosts. The first host-count accounts pin one account to each host so small oracle worlds still exercise the full topology.

func (*World) RelayAccountFloor 

func (w *World) RelayAccountFloor(pdsIndex int) int

RelayAccountFloor returns the incomplete count advertised by listHosts.

func (*World) RelayKnowsAccount 

func (w *World) RelayKnowsAccount(accountIdx int) bool

RelayKnowsAccount models the recreated relay's incomplete roster. Every third account is absent, guaranteeing a relay gap while preserving a large realistic subset for legacy relay-listRepos adversity tests.

func (*World) RelayListReposPage 

func (w *World) RelayListReposPage(start, limit int) ([]ListReposEntry, int, error)

RelayListReposPage pages only the relay-known subset.

func (*World) RepoUnavailableStatus 

func (w *World) RepoUnavailableStatus(idx int) (string, bool, error)

RepoUnavailableStatus returns the terminal getRepo-unavailable status for account idx, if one has been configured.

func (*World) RunTraffic 

func (w *World) RunTraffic(ctx context.Context, logger *slog.Logger) error

RunTraffic blocks generating + broadcasting events until ctx is cancelled. One event per loop iteration; inter-arrival drawn from the exponential distribution. Returns nil on graceful cancel.

func (*World) SetRepoUnavailableForTest 

func (w *World) SetRepoUnavailableForTest(idx int, status string) error

SetRepoUnavailableForTest makes getRepo for account idx return a terminal unavailable XRPC error. Status must be "takendown", "suspended", or "deactivated".

func (*World) SubscribeFanout 

func (w *World) SubscribeFanout() *fanout.Subscriber

SubscribeFanout adds a new subscriber to the live broadcast.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.