journal

package
v0.17.10 Latest Latest
Warning

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

Go to latest
Published: Oct 15, 2026 License: MIT Imports: 12 Imported by: 0

Documentation

Overview

Package journal is the daemon-owned durable event journal (ADR-007 D6, plan R-JRN): a single daemon-wide append-only log of versioned records (schema_version, monotonic cursor, ts, session_id, type, payload) under <stateDir>/journal/, written at the daemon's saveMetaLocked choke point (A7). It survives a daemon crash/upgrade (D-5): every append is fsync'd before its cursor is acked (R-JRN.5). Resume is "snapshot as of cursor N + events after N", atomic under one lock (R-JRN.4); a read below the retained floor returns a full-resync signal (R-JRN.6). Retention rotates segments to bound disk.

Index

Constants

View Source
const SchemaVersion = 1

SchemaVersion is the record schema this build writes. A future version is rejected loudly on decode; an older version is migrated forward (R-JRN.1).

Variables

This section is empty.

Functions

func EncodeRecord

func EncodeRecord(r Record) []byte

EncodeRecord serializes a record to one JSON line (no trailing newline).

Types

type Debouncer

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

Debouncer is the DELIVERY-layer flap coalescer (plan R-JRN.3, amendment D.0-A7): it NEVER lives in the durable journal (the durable log records every distinct net transition). A group_transition flap that settles within the window collapses to its final net state; a terminal lifecycle record is never delayed. A per-session window is anchored at the first offer for that session so a burst of flaps is coalesced and delivered once the burst settles.

func NewDebouncer

func NewDebouncer(window time.Duration, clock func() time.Time) *Debouncer

NewDebouncer builds a delivery-layer debouncer with the given window and clock.

func (*Debouncer) Drain

func (d *Debouncer) Drain(now time.Time) []Record

Drain returns every record DUE for delivery as of now: all immediate (terminal) records plus each session's coalesced group transition whose window has elapsed.

func (*Debouncer) Offer

func (d *Debouncer) Offer(r Record)

Offer enqueues a record at clock-now. A group_transition coalesces into its session's pending window (keeping the first offer's deadline, updating to the latest net state); every other record type is a terminal lifecycle record and is queued for immediate delivery, never debounced.

type Journal

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

Journal is the daemon-wide append-only log. Retained records are mirrored in memory (bounded by retention) so ReadFrom is atomic under one lock.

func Open

func Open(dir string, opts Options) (*Journal, error)

Open opens (or creates) the journal at dir, loading every retained record (tolerating a torn final line from a crash mid-append) and resuming the high-water cursor. Zero Options mean unbounded retention. A fresh active segment is always started so new appends never extend a possibly torn segment.

func (*Journal) Append

func (j *Journal) Append(r Record) (Record, error)

Append assigns the next cursor, stamps schema+ts, writes the record as one JSON line, and fsyncs BEFORE returning the stamped record (fsync-before-ack, R-JRN.5).

func (*Journal) Close

func (j *Journal) Close() error

Close closes the active segment file. Every append was already fsync'd, so a dropped handle (no Close) loses nothing.

func (*Journal) Cursor

func (j *Journal) Cursor() uint64

Cursor returns the current high-water cursor.

func (*Journal) ReadFrom

func (j *Journal) ReadFrom(from uint64) (Resume, error)

ReadFrom returns every retained record with cursor > from, atomically with the boundary cursor (R-JRN.4). If from fell below the retained floor (records were compacted away), FullResync is set so the caller resyncs rather than seeing a silent gap (R-JRN.6).

func (*Journal) Subscribe

func (j *Journal) Subscribe() (<-chan Record, func())

Subscribe registers a live subscriber and returns its record feed plus a cancel func. Every Record appended after subscription is delivered to the feed (subject to the non-blocking drop-on-full discipline in Append). The cancel func unsubscribes and closes the feed; it is idempotent and race-free (serialized with Append and Close under the journal lock).

func (*Journal) SubscribeFrom added in v0.13.19

func (j *Journal) SubscribeFrom(from uint64) (Resume, <-chan Record, func(), error)

SubscribeFrom atomically captures the retained backlog after from and registers the live feed that starts after the returned Cursor. No append can fall between those two actions: both occur under j.mu. The caller must apply res.Events before consuming live; appends during that work are bounded in live and overflow is signalled by closing it.

type Options

type Options struct {
	MaxBytes int64 // rotate the active segment past this size (0 = unbounded)
	MaxFiles int   // keep at most this many segments (0 = unbounded)
	// MaxTotalBytes drops the oldest whole segments at Open, before loading, until the rest
	// total at most this (0 = unbounded). Retained records are mirrored in memory.
	MaxTotalBytes int64
	Clock         func() time.Time // record timestamp source (defaults to time.Now)
}

Options tune retention (rotation bounds) and the injected clock.

type Record

type Record struct {
	SchemaVersion int          `json:"schema_version"`
	Cursor        uint64       `json:"cursor"`
	TS            time.Time    `json:"ts"`
	SessionID     string       `json:"session_id"`
	Type          RecordType   `json:"type"`
	Group         status.Group `json:"group,omitempty"` // set on group_transition
	// Agent is the session's agent identity (persist.Meta.AgentType), so the phone can label
	// a session with the CLI running it. The daemon sets it on the roster snapshot and on all
	// four journalworthy transitions (internal/daemon/journal.go); the `deleted` record
	// (lifecycle.go) deliberately carries none, because it REMOVES the session from the
	// phone's model and an agent name on a tombstone would describe nothing.
	//
	// It is omitempty and its ABSENCE IS MEANINGFUL: a record with no agent carries no
	// agent, and readers must never read "" as an agent by that name.
	//
	// Adding it did NOT bump SchemaVersion, and internal/journal/agentfield_test.go pins
	// why in both directions: this build reads pre-Agent lines (an absent key decodes to
	// ""), and a build predating the field reads records carrying one (encoding/json
	// ignores the unknown key). A bump would instead make every older daemon reject every
	// record this build writes, costing them the whole journal to gain nothing.
	Agent string `json:"agent,omitempty"`
	// Name is the session's USER-GIVEN label (persist.Meta.Name), so the phone can head a row
	// with what a person called the session instead of the 16-char random local id. It is set
	// at exactly the sites Agent is -- the roster snapshot and all four journalworthy
	// transitions -- and the `deleted` record deliberately carries none, for Agent's reason: a
	// tombstone REMOVES the session, and a label on it would describe nothing.
	//
	// It is omitempty and its ABSENCE IS MEANINGFUL, again as Agent's: most sessions are never
	// labelled, so an empty name means the user gave none. The phone falls back to the id
	// (mobile/app.go's session()); nothing anywhere may substitute a plausible label for a
	// missing one.
	//
	// Adding it did NOT bump SchemaVersion, and internal/journal/namefield_test.go pins why in
	// both directions, exactly as agentfield_test.go did for Agent.
	Name string `json:"name,omitempty"`
	// StateSince is the MACHINE's own stamp of when the session entered its CURRENT state --
	// persist.Meta.EffectiveGroupEnteredAt(), with that method's own fallbacks for records
	// written before GroupEnteredAt existed -- set at exactly the sites Agent and Name are
	// (the roster snapshot and the four journalworthy transitions), so the phone can age a
	// row from a clock it did not invent (phone-refit-playbook W7.1). It is NOT LastActivity:
	// that is written only at launch and exit, so "Waiting on you · 3h" would have meant
	// "launched 3h ago" and twins launched together would age identically. Zero means the
	// meta carried no instant at all, and it is omitzero so a record without one serialises
	// as it did before the field existed. It is distinct from TS, which is when THIS RECORD
	// was appended.
	StateSince time.Time       `json:"state_since,omitzero"`
	Payload    json.RawMessage `json:"payload,omitempty"` // opaque per-type extra
}

Record is one versioned journal entry. Cursor is a monotonic uint64 assigned by Append and never reused for the journal's whole lifetime.

func DecodeRecord

func DecodeRecord(line []byte) (Record, error)

DecodeRecord parses one JSON line. A future schema version is rejected loudly (never silently dropped); an older version is migrated forward; malformed JSON is an error (R-JRN.1).

type RecordType

type RecordType string

RecordType is the kind of journalled transition.

const (
	TypeGroupTransition RecordType = "group_transition"
	TypeLaunched        RecordType = "launched"
	TypeExited          RecordType = "exited"
	TypeLost            RecordType = "lost"
	TypeDeleted         RecordType = "deleted"
	TypePresence        RecordType = "presence"
	// TypeInteraction is one interaction item -- a message, tool run, file change,
	// approval, plan revision or status marker of the phone's chat transcript (ADR-009,
	// ADR-010, docs/specifications/interaction-schema.md IS-LAYER-1). The item object IS
	// the record's Payload, and the item's own `kind` discriminator lives only in there:
	// nothing that spec defines may become a record field, let alone a header field the
	// seq bucket keys on (IS-LAYER-2, PB-SYNC-1). Ordering is the cursor Append assigns and
	// nothing else -- an item carries no private sequence number (IS-LAYER-3).
	//
	// Adding the type did NOT bump SchemaVersion, and interaction_test.go pins why: a new
	// TYPE is a new value of an existing string field, so the on-disk field set is
	// unchanged, and a reader that does not know the type skips the record and advances its
	// cursor (IS-COMPAT-1/-3) instead of rejecting the whole journal the way a version bump
	// would make it.
	TypeInteraction RecordType = "interaction"
	// TypeRoster is a SYNTHETIC snapshot record: it is emitted ONLY inside
	// Resume.Roster (the atomic snapshot half of R-JRN.4) to describe a session that
	// is live as-of the read cursor. It is NEVER appended to the journal — no
	// producer writes it and no reader will find it in the event stream. The daemon
	// (which owns the meta store) populates these; the journal package itself never
	// manufactures a roster record.
	TypeRoster RecordType = "roster"
	// TypeStructuredGap is the daemon-authored capability-degrade event of ADR-017 T2
	// rule 2: an unrecoverable shim/daemon spool or cursor gap emits an exact boundary,
	// disables structured_chat for the session instance, and forbids a fabricated
	// completion. Its body is daemon.StructuredGapEvent, riding this record family exactly
	// as InteractionItem rides TypeInteraction -- no new mailbox frame kind.
	TypeStructuredGap RecordType = "structured_gap"
	// TypeCapabilityTransition is an additive, cursor-ordered publication of the
	// daemon's complete capability record after a durable runtime transition. It is
	// separate from TypeRoster: roster records are snapshot-only, while a phone that
	// is already subscribed must observe a degrade or a proven recovery without a
	// reconnect. The payload is protocol.SessionCapabilities; journal deliberately
	// owns only the string discriminator to avoid an import cycle.
	TypeCapabilityTransition RecordType = "capability_transition"
	// TypeSessionState is an additive authoritative state delta for one live
	// session. Unlike TypeRoster it is durable/cursor ordered, and unlike
	// TypeCapabilityTransition it may establish a new exact session instance. Its
	// payload, when present, is the complete validated SessionCapabilities value at
	// this cursor.
	TypeSessionState RecordType = "session_state"
)

type Resume

type Resume struct {
	Cursor     uint64
	Roster     []Record
	Events     []Record
	FullResync bool
}

Resume is the atomic result of ReadFrom: the snapshot roster as of the read cursor, the events after the caller's cursor, and a full-resync signal when the caller's cursor fell below the retained floor (dropped records were not silently omitted, R-JRN.6).

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL