Documentation
¶
Overview ¶
Package checkpoint records signed statements about where a hash chain stood, so that rewriting an event after the fact becomes provable.
A chain of unkeyed digests can be recomputed by anyone who can write to the store. Keying the digest raises that bar, but both the event's scheme and its stream's pin live in the same database, so an attacker with write access can still rewrite both. A checkpoint is different: it asserts that at sequence N the chain hash was H, and it is signed, so the assertion cannot be restated without the signing key. If the event now at N hashes to something else, the chain was rewritten after the checkpoint was taken and the signature proves it.
Index ¶
Constants ¶
const AlgorithmEd25519 = "ed25519"
AlgorithmEd25519 names the signature algorithm recorded on a checkpoint.
Variables ¶
var ( // ErrNotFound is returned when a checkpoint cannot be found. ErrNotFound = errors.New("checkpoint: not found") // ErrExists is returned when a checkpoint already covers a stream's // sequence: another one already ends at the same to_seq, or another one // already starts at the same from_seq. // // Two checkpointers racing on one stream is expected: a background ticker // and an operator-triggered run can overlap, as can two replicas' tickers. // Rather than rely on lock discipline, which only ever covers one // process, the store makes an overlapping checkpoint structurally // impossible and the loser of the race gets this. It is an ordinary // outcome, not a fault: the scheduler treats it as one. ErrExists = errors.New("checkpoint: already exists for this sequence") // ErrUnsupported is returned by backends that cannot store checkpoints // durably enough to be a root of trust. ErrUnsupported = errors.New("checkpoint: this store does not support checkpoints") )
Sentinel errors.
These live here rather than in the root package's errors.go, next to ErrEventNotFound and its neighbors, because checkpoint must not import chronicle: chronicle imports verify, stream imports chronicle, and a later task has verify import checkpoint. An import back to chronicle from here would close that into a cycle.
var ErrNothingToCheckpoint = errors.New("checkpoint: no new events since the last checkpoint")
ErrNothingToCheckpoint is returned when a stream has gained no events since its last checkpoint. It is an ordinary outcome on a quiet stream, not a fault.
Functions ¶
func CanonicalPayload ¶
func CanonicalPayload(c *Checkpoint) string
CanonicalPayload renders the bytes a checkpoint's signature covers.
Every field the checkpoint asserts appears here. A field left out is a field an attacker can edit while the signature still verifies.
Variable-length string fields are rendered length-prefixed (lengthPrefixed: "<byte-length>:<value>") rather than joined with a bare "|". AppID and TenantID are free text from caller scope, so a bare join lets a "|" inside one value shift where the reader thinks the next field starts: AppID="a", TenantID="b|c" and AppID="a|b", TenantID="c" would otherwise render identically. Length-prefixing fixes each field's end at an exact byte offset instead of at the next "|", so no embedded separator can move a boundary. FromHash, ToHash, PrevCheckpoint, and StreamID get the same treatment on the same reasoning, even though today's callers only produce hex and TypeID strings for them. FromSeq, ToSeq, EventCount, and CreatedAt are left as plain decimal/RFC3339Nano text: none of those formats can contain "|", so they cannot shift a boundary either way.
Types ¶
type Checkpoint ¶
type Checkpoint struct {
ID id.ID `json:"id"`
StreamID id.ID `json:"stream_id"`
AppID string `json:"app_id"`
TenantID string `json:"tenant_id,omitempty"`
// FromSeq and ToSeq bound the range this checkpoint covers, inclusive.
FromSeq uint64 `json:"from_seq"`
ToSeq uint64 `json:"to_seq"`
// FromHash and ToHash are the chain hashes at those boundaries. ToHash is
// the assertion that matters: recomputing the event at ToSeq must reproduce
// it, or the chain was rewritten after this checkpoint.
FromHash string `json:"from_hash"`
ToHash string `json:"to_hash"`
// EventCount is the sequence span this checkpoint covers (ToSeq - FromSeq
// + 1), not a verified count of rows the store actually holds in that
// range. CheckpointStream computes it arithmetically from the two
// sequence numbers it already asserts, without reading a single event;
// a gap in the range (a missing sequence) is not reflected here and is
// verification's job (Store.Gaps), not a checkpoint's.
EventCount int64 `json:"event_count"`
// PrevCheckpoint is the digest of the preceding checkpoint's signed payload,
// which is what makes deleting one from the middle detectable. It is empty
// for a stream's first checkpoint.
PrevCheckpoint string `json:"prev_checkpoint,omitempty"`
Algorithm string `json:"algorithm"`
SignKeyID string `json:"sign_key_id"`
Signature []byte `json:"signature"`
// SignedPayload is the exact bytes that were signed. Storing them rather
// than re-deriving from the columns means a later change to the canonical
// encoding cannot silently invalidate every historical signature.
SignedPayload string `json:"signed_payload"`
CreatedAt time.Time `json:"created_at"`
}
Checkpoint is a signed statement about one stream's chain over a sequence range.
type Checkpointer ¶
type Checkpointer struct {
// contains filtered or unexported fields
}
Checkpointer takes signed checkpoints over a stream's sequence range.
It holds its own per-stream lock rather than sharing Chronicle's. Sharing would make checkpointing block event writes, and that lock is already the per-tenant write ceiling. A checkpoint reads the head at some instant and signs that; concurrent appends simply land in the next window.
func NewCheckpointer ¶
func NewCheckpointer(store Store, signer Signer, logger log.Logger, opts ...CheckpointerOption) *Checkpointer
NewCheckpointer creates a Checkpointer. A nil logger is replaced with a no-op.
It does not take an EventReader: CheckpointStream never reads an event, so there is nothing here for one to do. See CheckpointStream's doc for why.
func (*Checkpointer) CheckpointStream ¶
func (c *Checkpointer) CheckpointStream(ctx context.Context, st StreamHead) (*Checkpoint, error)
CheckpointStream signs the stream's current head state and stores the result.
The window runs from one past the previous checkpoint's ToSeq (or 1 for a stream's first) up to the stream's head.
It does not read a single event to do this. Everything a checkpoint asserts -- FromSeq, ToSeq, EventCount -- is arithmetic on sequence numbers the caller already supplied (st.HeadSeq) or this function already read (latest.ToSeq); reading and counting the actual event rows would add nothing a verifier trusts, since ToHash already pins the head and a gap in between is Store.Gaps' job to find, not a checkpoint's. Before this, materialising every event in [fromSeq, st.HeadSeq] just to call len() on the result meant a first checkpoint over a stream with millions of existing events read that entire history into memory, serially, per stream, in one goroutine -- and runCheckpointScheduler is what put this on an automatic, per-stream, per-tick path across an entire deployment.
Each call is one "take checkpoint" span. A stream with nothing new and a checkpoint another writer took first are ordinary outcomes: the span records them in chronicle.checkpoint.result and keeps an unset status.
type CheckpointerOption ¶
type CheckpointerOption func(*Checkpointer)
CheckpointerOption configures a Checkpointer.
func WithMetrics ¶
func WithMetrics(m gometrics.Metrics) CheckpointerOption
WithMetrics counts every CheckpointStream call on m by result (ok, nothing_new, lost_race, error). In a Forge app pass app.Metrics(), and the series lands on /_/metrics as forge_chronicle_checkpoints_attempted_total. The Forge extension does this for you. A nil m records nothing.
type Ed25519Signer ¶
type Ed25519Signer struct {
// contains filtered or unexported fields
}
Ed25519Signer signs checkpoints with an ed25519 key from a keys.Provider.
Verification resolves the key through the provider by ID, so revoking a signing key invalidates every checkpoint it signed. That is deliberate and matches how revoking an hmac key invalidates the events it digested.
func NewEd25519Signer ¶
func NewEd25519Signer(provider keys.Provider) *Ed25519Signer
NewEd25519Signer creates a signer backed by the given provider.
type Signer ¶
type Signer interface {
// Sign returns the signature, the ID of the key used, and the algorithm.
Sign(ctx context.Context, payload []byte) (sig []byte, keyID, alg string, err error)
// Verify checks a signature against the key that made it.
Verify(ctx context.Context, payload, sig []byte, keyID string) error
// PublicKey returns the public half of a signing key, so an auditor can
// check a checkpoint without holding anything secret.
PublicKey(ctx context.Context, keyID string) ([]byte, error)
}
Signer signs and verifies checkpoint payloads.
type Store ¶
type Store interface {
// AppendCheckpoint persists a checkpoint. It returns ErrExists if one
// already covers the same (stream, to_seq).
AppendCheckpoint(ctx context.Context, cp *Checkpoint) error
// LatestCheckpoint returns the checkpoint with the highest ToSeq for a
// stream, or ErrNotFound when the stream has none.
LatestCheckpoint(ctx context.Context, streamID id.ID) (*Checkpoint, error)
// CheckpointsInRange returns every checkpoint whose range OVERLAPS
// [fromSeq, toSeq], ascending by ToSeq.
//
// Overlap rather than containment, because a caller verifying sequences
// 150 to 160 still needs the checkpoint covering 101 to 200: that is the
// one whose assertion those events fall under.
CheckpointsInRange(ctx context.Context, streamID id.ID, fromSeq, toSeq uint64) ([]*Checkpoint, error)
// GetCheckpoint returns one checkpoint by ID.
GetCheckpoint(ctx context.Context, checkpointID id.ID) (*Checkpoint, error)
// ListCheckpoints returns a stream's checkpoints, newest first.
ListCheckpoints(ctx context.Context, streamID id.ID, opts ListOpts) ([]*Checkpoint, error)
}
Store persists signed checkpoints.
Checkpoints are append-only. There is no update and no delete: a checkpoint that could be revised would assert nothing.
type StreamHead ¶
StreamHead is the minimal view of a stream a checkpoint is taken over.
It exists so this package does not import stream, which imports chronicle, which imports verify, which imports this package. chronicle.StreamInfo is the same device for the same reason; callers convert at the boundary.
Source Files