README
¶
pstate
pstate is a small, dependency-free persistent JSON state store for scheduled jobs and services. It began as the immutable state core in prangl and is designed for independent angl heartbeat invocations that need to share durable, queryable state without operating a database server.
- Untyped: keys are strings; values are arbitrary JSON.
- Deterministic: values are compacted, object keys and query results are ordered, and queries have explicit JSON semantics.
- Durable: writes use same-directory temporary files, file flushes, atomic replacement, and directory flushes on Unix.
- Process-safe: each transaction locks a sidecar
<store>.lock, reloads the latest revision, and commits once. Concurrent jobs do not overwrite one another. - Snapshot-friendly: the Go package includes an immutable AVL tree with structural sharing and a lock-free CAS model whose background writer coalesces snapshots.
- Portable: Windows and Unix implementations use only the Go standard library.
Install
go install github.com/jack-work/pstate/cmd/pstate@latest
For a Go module:
go get github.com/jack-work/pstate@latest
CLI
Choose a state file explicitly for each job, or set PSTATE_FILE. If neither is set, the default is <user-config>/pstate/state.json.
$state = "$env:LOCALAPPDATA\pstate\deployments.json"
pstate --file $state set jobs/orchard '{"state":"running","attempts":2,"owner":"orchard"}'
pstate --file $state get jobs/orchard
pstate --file $state query 'key ^= "jobs/" && .state == "running"'
pstate --file $state query --keys '.attempts >= 2'
pstate --file $state delete jobs/orchard
get emits the JSON value. query emits a JSON array of {key,value} objects sorted by key. Use --jsonl for one object per line or --keys for plain keys. Successful mutations are quiet, making them safe in angl logs. Diagnostics go to stderr.
Apply several edits in one durable transaction:
@'
[
{"op":"set","key":"jobs/a","value":{"state":"ready"}},
{"op":"set","key":"jobs/b","value":[1,2,3]},
{"op":"delete","key":"jobs/old"}
]
'@ | pstate --file $state apply -
Exit statuses are 0 for success, 1 for operational/query errors, 2 for usage errors, and 3 when get cannot find its key.
Run pstate --help for the complete command guide.
Query language
The query language is deliberately small and deterministic:
key ^= "jobs/" && .state == "failed"
.attempts >= 3 && (.owner == "orchard" || !.owner)
.runs[0].status == "timed-out"
value["odd key"] *= "needle"
Selectors
| Selector | Meaning |
|---|---|
key |
Entry key as a string |
value |
Entire JSON value |
.field, value.field |
Object field |
.runs[0] |
Array index |
value["odd key"] |
Quoted object field |
A bare selector tests existence. Missing paths are false for every comparison, including !=; therefore !.owner means “owner is absent.”
Operators
| Operators | Types / meaning |
|---|---|
==, != |
JSON equality; numbers compare by exact mathematical value (3 == 3.0) |
<, <=, >, >= |
Exact numbers or ordinal strings; mixed types do not match |
^=, $=, *= |
String prefix, suffix, or substring |
!, &&, ||, ( ) |
Negation, short-circuit conjunction/disjunction, grouping |
Literals use JSON syntax. There are no timestamps, environment interpolation, regex engines, locale-dependent comparisons, or implicit string/number conversions.
Go package
Use Store when separate processes may touch the same file:
package main
import (
"log"
"github.com/jack-work/pstate"
)
func main() {
store, err := pstate.Open(`C:\state\job.json`)
if err != nil { log.Fatal(err) }
value, err := pstate.EncodeValue(map[string]any{
"state": "ready",
"attempts": 1,
})
if err != nil { log.Fatal(err) }
snapshot, err := store.Apply(pstate.Patch{}.
Set("jobs/a", value).
Delete("jobs/old"))
if err != nil { log.Fatal(err) }
log.Printf("revision %d contains %d keys", snapshot.Version(), snapshot.Len())
}
Tree and Snapshot are immutable and safe to retain while newer roots are published. Model is for one long-lived process that wants lock-free reads and coalesced asynchronous serialization:
model := pstate.NewModel(initialTree, pstate.WriterFunc(func(s pstate.Snapshot) error {
// Serialize this immutable point-in-time snapshot.
return nil
}))
defer model.Close()
A Model coordinates goroutines in one process; a Store coordinates transactions between processes. Do not use an independently loaded Model as a multi-process read-modify-write mechanism.
File format and recovery
The store is human-readable JSON:
{
"schema": 1,
"revision": 2,
"entries": [
{"key": "jobs/a", "value": {"state": "ready"}}
]
}
Entries are serialized in lexical key order. A mutation that makes no net change does not rewrite the file or increment revision. The .lock file is intentionally persistent and contains no state. A failed or interrupted replacement leaves the preceding complete snapshot readable; temporary files may be removed safely when no writer is active.
Development
go test ./...
go test -race ./...
go vet ./...
The test suite covers AVL invariants and structural sharing, immutable snapshots, CAS concurrency, writer coalescing and failures, durable file round trips, overlapping transactions, query parsing/evaluation, and CLI streams/exit statuses.
License
MIT
Documentation
¶
Overview ¶
Package pstate provides a schema-free persistent JSON state store for jobs and services. It combines immutable ordered snapshots, lock-free in-process models, and process-safe atomic file transactions.
Index ¶
- Constants
- Variables
- type Change
- type Entry
- type Follower
- type Model
- func (m *Model) Apply(patch Patch) (Snapshot, error)
- func (m *Model) Close() error
- func (m *Model) Delete(key string) (Snapshot, error)
- func (m *Model) Flush() error
- func (m *Model) Get(key string) (Value, bool)
- func (m *Model) LastError() error
- func (m *Model) Len() int
- func (m *Model) Set(key string, value Value) (Snapshot, error)
- func (m *Model) Snapshot() Snapshot
- type Patch
- type Position
- type Query
- type Snapshot
- type Store
- func (s *Store) Apply(patch Patch) (Snapshot, error)
- func (s *Store) Delete(key string) (Snapshot, error)
- func (s *Store) Follow(from Position) (*Follower, error)
- func (s *Store) Get(key string) (Value, bool, error)
- func (s *Store) Path() string
- func (s *Store) Read() (Snapshot, error)
- func (s *Store) ReadAt() (Snapshot, Position, error)
- func (s *Store) Set(key string, value Value) (Snapshot, error)
- type Tree
- type Value
- type Writer
- type WriterFunc
Constants ¶
const Wire = 1
Wire is the version of the change-stream protocol: the journal record, the Position, and their JSON encodings. It is NOT the store's file format (see fileSchema) and it is NOT a revision.
ONE WORD, THREE QUESTIONS, AND THEY ARE KEPT APART ON PURPOSE. "Version" in this package could mean the shape of the bytes, the shape of the protocol, or how many times the state has changed. Snapshot.Version() answers the third and nothing else. A client compares THIS number to decide whether it can interpret a record at all, and a client that finds a record it cannot interpret must fall back to snapshot-only rather than skip the record -- which is the whole reason the number is on the wire instead of only in go.mod.
Variables ¶
var ( // ErrGap reports that the stream cannot be proved continuous from the follower's position. It is // not a warning and it must not be logged and ignored: the only sound response is to take a // fresh snapshot and follow again from its Position. ErrGap = errors.New("pstate: change stream gap") // ErrWire reports a record this build cannot interpret. A follower that sees it should degrade // to snapshot-only. It is separate from ErrGap because they call for different things: a gap is // recoverable by re-snapshotting and continuing to follow, a wire mismatch is not. ErrWire = errors.New("pstate: unsupported wire version") // ErrJournal reports that the state was COMMITTED but the change record was not appended. It is // its own sentinel because the alternatives are both lies: returning a plain error tells the // caller its write failed when the write landed, and returning nil tells a follower nothing // while the stream quietly loses a record. errors.Is(err, ErrJournal) means: your data is safe, // the stream is not, and every follower will detect the gap. ErrJournal = errors.New("pstate: change not journalled") )
var ErrClosed = errors.New("pstate: model closed")
ErrClosed is returned by operations that cannot run after Close begins.
Functions ¶
This section is empty.
Types ¶
type Change ¶ added in v0.2.0
type Change struct {
Wire int `json:"wire"`
Epoch string `json:"epoch"`
From uint64 `json:"from"`
To uint64 `json:"to"`
Patch Patch `json:"edits"`
}
Change is one committed transaction: the patch, and the two revisions it spans.
From is what makes the stream provable rather than merely ordered. A follower applies a change only when From equals the revision it currently holds, so "the next record" is decided by the record itself and never by the order it happened to be read in. Anything else is a gap.
type Follower ¶ added in v0.2.0
type Follower struct {
// contains filtered or unexported fields
}
Follower reads a store's change stream from a Position.
IT NEVER TAKES THE STORE LOCK AND IT NEVER OPENS THE STORE FILE. Both are deliberate. pstate's lock QUEUES -- LockFileEx with no LOCKFILE_FAIL_IMMEDIATELY, so a second writer waits rather than standing down -- which is the right policy for a store whose callers expect their write to land, and it is exactly why a reader must stay outside it: a follower that queued for the lock would add its own read latency to every writer on the box, and one that stalled would stall them all. The journal is a separate append-only sidecar opened with FILE_SHARE_DELETE, so a follower can neither block the writer's atomic replace nor be blocked by it.
func (*Follower) Close ¶ added in v0.2.0
Close releases the follower. It holds no handle between calls, so this only bars further reads.
func (*Follower) Next ¶ added in v0.2.0
Next returns the next change, or false when the follower has read everything committed so far.
It advances the follower's Position only on a change it could prove was next. ErrGap leaves the position untouched, so a caller that re-snapshots does not have to reason about how far it got.
type Model ¶
type Model struct {
// contains filtered or unexported fields
}
Model provides lock-free snapshots and CAS updates. Its single background worker coalesces notifications and calls Writer serially. Use Store for coordination between separate processes.
func NewModel ¶
NewModel starts a model containing initial. A nil writer disables I/O while retaining Flush and Close semantics.
func (*Model) Apply ¶
Apply atomically publishes all edits as one update. A patch with no net change does not advance the version.
func (*Model) Close ¶
Close rejects future updates, flushes the final snapshot, and stops the worker.
func (*Model) Flush ¶
Flush waits until a snapshot at least as recent as the one captured at call start has been persisted successfully.
func (*Model) LastError ¶
LastError returns the most recent background persistence error. A successful write clears it.
type Patch ¶
type Patch struct {
// contains filtered or unexported fields
}
Patch is an immutable ordered batch of edits. Its zero value is empty. If a key occurs more than once, the final edit wins.
func (Patch) MarshalJSON ¶ added in v0.2.0
MarshalJSON encodes the patch as an ordered array of edits. Superseded edits are PRESERVED rather than collapsed: a patch is an ordered batch by definition, Tree.Apply replays it in order, and a reader that folds the encoded form must reach the same tree as one that folded the original. A "helpful" collapse here would make those two disagree only for patches that touch a key twice.
func (*Patch) UnmarshalJSON ¶ added in v0.2.0
UnmarshalJSON decodes an ordered array of edits.
STRICT, and deliberately so: an unknown field is an error rather than a shrug. A decoder that drops what it cannot name turns a newer writer's record into a silently smaller one, and the fold that results is wrong without being detectably wrong. Version skew is the wire version's job -- it is checked before this is ever reached -- so an unknown field HERE means a defect and not a vintage.
type Position ¶ added in v0.2.0
type Position struct {
Wire int `json:"wire"`
Epoch string `json:"epoch"`
Revision uint64 `json:"revision"`
}
Position is where a reader stands in a store's change stream: which protocol it speaks, which incarnation of the store it is reading, and how far it has folded.
A BARE REVISION IS NOT A POSITION, AND THIS TYPE EXISTS TO SAY SO. A revision counts net changes within one incarnation of one store. Delete the state file and the next write starts counting again from a number the client has already seen, so two different states share one label and nothing in the integer can tell them apart. Epoch is what makes the pair unique: it is minted when a store first commits and it is carried in the state file, so it survives every process that writes and changes exactly when the state itself was replaced rather than advanced.
type Query ¶
type Query struct {
// contains filtered or unexported fields
}
Query is a compiled deterministic predicate over an entry. Queries are safe for concurrent use.
func ParseQuery ¶
ParseQuery compiles an expression. An empty expression matches every entry.
Selectors are key, value, .field, value.field, and array/object paths such as .runs[0].status or value["odd key"]. Operators are ==, !=, <, <=, >, >=, ^= (prefix), $= (suffix), and *= (substring). Combine predicates with !, &&, ||, and parentheses. Literals use JSON syntax.
type Snapshot ¶
type Snapshot struct {
// contains filtered or unexported fields
}
Snapshot is an immutable point-in-time view. Version increases once per successfully published Model update or Store transaction.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is a process-safe file-backed state store. Each mutation takes an exclusive sidecar lock, reloads the latest snapshot, and durably replaces the file, preventing lost updates between independent heartbeat invocations.
func (*Store) Apply ¶
Apply atomically commits a patch against the latest on-disk snapshot. A patch with no net change does not rewrite the file or advance its revision.
A COMMIT THAT CHANGES NOTHING PUBLISHES NOTHING, and that is what makes the revision a usable stream position rather than a write counter: the sequence a follower folds is exactly the sequence of states the store has been in. Break this and every "provably next" claim in journal.go weakens to "next by arrival order".
On success the change is appended to the journal. If the state commits but the journal does not, the returned error wraps ErrJournal and the returned snapshot is the committed one: the caller's data is safe, and every follower will take ErrGap and re-snapshot rather than miss the change.
func (*Store) Follow ¶ added in v0.2.0
Follow returns a follower positioned immediately after from.
It scans the journal for the record that continues from, which costs one bounded read of a file capped at journalMaxBytes. If the stream cannot be proved to continue from that position -- the records are from another epoch, or the history has been discarded, or from is ahead of the journal -- it returns ErrGap and the caller must take a fresh snapshot rather than start reading anyway.
func (*Store) ReadAt ¶ added in v0.2.0
ReadAt returns a consistent immutable snapshot AND the point in the change stream it was taken at.
A SNAPSHOT WITHOUT ITS POSITION CANNOT BE THE BASELINE OF A PATCH STREAM. A reader that loads state and then applies changes to it is asserting that the first change it applies is the one that immediately followed the state it loaded, and nothing in a bare Snapshot can support that claim.
THE READ PATH TAKES NO LOCK, AND THAT IS A CORRECTNESS ARGUMENT RATHER THAN A SHORTCUT. It used to take the same LockFileEx the writers take -- flag 0x2, EXCLUSIVE, with no LOCKFILE_FAIL_IMMEDIATELY and no deadline. King g3 measured what that costs: 11.06 seconds of contended wait inside Open(), success 99ms after the holder released, negative control 0.00s. A reader therefore serialised against every writer AND every other reader, unbounded, with no error for a caller to handle, and it did so at CONSTRUCTION -- before any caller could hold anything.
The lock bought nothing on this path. writeSnapshot ends in an atomic replace, so the file a reader opens is always a whole revision: it sees the state before that replace or the state after it, never a mixture, because the bytes of a revision are assembled in a temporary file and swapped in as one operation. That is the same property the lock was being asked to provide, and the file already had it. What the lock added was the ability for a reader to make a writer wait, which is the exact inversion a follower exists to avoid.
So the pair below is consistent for the same reason a single read is: it reads one file once. Store.Apply still takes the lock, because a read-modify-write genuinely needs one.
A reader can still momentarily fail a writer's MoveFileExW by holding the destination open -- that is measured and unavoidable -- which is why readShared grants FILE_SHARE_DELETE and why replaceFile retries. Those two make a reader HARMLESS to a writer. Taking the writers' lock made it the opposite.
type Tree ¶
type Tree struct {
// contains filtered or unexported fields
}
Tree is an immutable, string-keyed ordered map of JSON values. Its zero value is empty. Set and Delete share all unaffected nodes with the original tree.
type Value ¶
type Value struct {
// contains filtered or unexported fields
}
Value is an immutable, canonical JSON value. Its zero value is JSON null.
func EncodeValue ¶
EncodeValue converts v to an immutable JSON value.
func ParseValue ¶
ParseValue validates data and returns its deterministic compact encoding.
func (Value) Equal ¶
Equal reports JSON equality. Values are canonical, so this is constant time apart from comparing their encodings.
func (Value) MarshalJSON ¶
MarshalJSON implements json.Marshaler.
func (*Value) UnmarshalJSON ¶
UnmarshalJSON implements json.Unmarshaler.
type WriterFunc ¶
WriterFunc adapts a function to Writer.
func (WriterFunc) WriteSnapshot ¶
func (f WriterFunc) WriteSnapshot(snapshot Snapshot) error
WriteSnapshot calls f(snapshot).
Source Files
Directories