journal

package
v0.13.12 Latest Latest
Warning

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

Go to latest
Published: Aug 31, 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) (*Journal, error)

Open opens (or creates) the journal at dir with default (unbounded) retention.

func OpenWithOptions

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

OpenWithOptions opens the journal, loading every retained record (tolerating a torn final line from a crash mid-append) and resuming the high-water cursor. 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).

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)
	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"
)

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