// Package logstorage implements the authoritative, bounded on-device event // journal owned by tera-logd. Closed segment files are immutable; the index is // derived metadata and may always be rebuilt from the checksummed records. package logstorage import ( "bufio" "bytes" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "io" "io/fs" "os" "path/filepath" "sort" "strings" "sync" "time" "tera-clone/services/tera-api/internal/storage/atomicfile" ) const ( ClassAudit = "audit" ClassSystem = "system" ClassPort = "port" ClassLink = "link" formatVersion = 1 defaultSegmentSize = int64(4 << 20) minimumSegmentSize = int64(1 << 20) maximumRecordBytes = int64(64 << 10) pageCheckpointRecords = 256 ) var ( ErrCapacityExhausted = errors.New("log storage capacity exhausted") ErrCorruptSegment = errors.New("log segment integrity check failed") errPageComplete = errors.New("log page complete") ) type Options struct { Root string SegmentBytes int64 AuditCapBytes int64 SystemCapBytes int64 PortCapBytes int64 IndexCapBytes int64 GlobalFreeFloor int64 Clock func() time.Time FreeBytes func(string) (int64, error) } type Retention struct { AuditDays int `json:"audit_days"` SystemDays int `json:"system_days"` PortDays int `json:"port_days"` } type Envelope struct { FormatVersion int `json:"format_version"` Sequence uint64 `json:"sequence"` Timestamp time.Time `json:"timestamp"` Class string `json:"class"` Payload json.RawMessage `json:"payload"` Checksum string `json:"checksum_sha256"` } type classState struct { NextSequence uint64 ActiveFirst time.Time ActiveLast time.Time ActiveFirstSequence uint64 ActiveLastSequence uint64 ActiveBytes int64 ActiveRecords int ActiveOffsets []recordOffset } type recordOffset struct { Sequence uint64 `json:"sequence"` Offset int64 `json:"offset"` } type segmentIndex struct { FormatVersion int `json:"format_version"` RebuiltAt time.Time `json:"rebuilt_at"` Classes map[string]uint64 `json:"next_sequence"` Segments []segmentMeta `json:"segments"` } type segmentMeta struct { Class string `json:"class"` Path string `json:"path"` Closed bool `json:"closed"` FirstSequence uint64 `json:"first_sequence"` LastSequence uint64 `json:"last_sequence"` Oldest time.Time `json:"oldest"` Newest time.Time `json:"newest"` Bytes int64 `json:"bytes"` Records int `json:"records"` PageOffsets []recordOffset `json:"page_offsets,omitempty"` } type Snapshot struct { Root string `json:"root"` FreeBytes int64 `json:"free_bytes"` GlobalFreeFloor int64 `json:"global_free_floor_bytes"` UsageBytes map[string]int64 `json:"usage_bytes"` CapBytes map[string]int64 `json:"cap_bytes"` NextSequence map[string]uint64 `json:"next_sequence"` Oldest map[string]time.Time `json:"oldest"` Newest map[string]time.Time `json:"newest"` } type Store struct { mu sync.Mutex root string segment int64 caps map[string]int64 indexCap int64 floor int64 clock func() time.Time freeBytes func(string) (int64, error) retention Retention classes map[string]*classState closed map[string][]segmentMeta portClears portClearState systemClear systemClearState linkClear linkClearState } func Open(options Options) (*Store, error) { root := filepath.Clean(strings.TrimSpace(options.Root)) if root == "." || !filepath.IsAbs(root) { return nil, errors.New("log storage root must be an absolute path") } info, err := os.Lstat(root) if err != nil { return nil, fmt.Errorf("inspect log storage root: %w", err) } if !info.IsDir() || info.Mode()&os.ModeSymlink != 0 { return nil, errors.New("log storage root must be a real directory") } segmentBytes := options.SegmentBytes if segmentBytes == 0 { segmentBytes = defaultSegmentSize } if segmentBytes < minimumSegmentSize || segmentBytes > defaultSegmentSize { return nil, fmt.Errorf("log segment size must be between %d and %d bytes", minimumSegmentSize, defaultSegmentSize) } clock := options.Clock if clock == nil { clock = time.Now } free := options.FreeBytes if free == nil { free = filesystemFreeBytes total, err := filesystemTotalBytes(root) if err != nil { return nil, err } // Reserve at least 10% of the filesystem: log capacity is fixed at 90%. if total/10 > options.GlobalFreeFloor { options.GlobalFreeFloor = total / 10 } } store := &Store{ root: root, segment: segmentBytes, indexCap: options.IndexCapBytes, floor: options.GlobalFreeFloor, clock: clock, freeBytes: free, retention: Retention{AuditDays: 365, SystemDays: 365, PortDays: 365}, caps: map[string]int64{ "audit": options.AuditCapBytes, "system": options.SystemCapBytes, "port": options.PortCapBytes, }, classes: map[string]*classState{}, closed: map[string][]segmentMeta{}, } for _, directory := range []string{"audit", "system", "port", "index", "export"} { path := filepath.Join(root, directory) if err := ensureDirectory(path, directory == "export"); err != nil { return nil, err } } store.mu.Lock() defer store.mu.Unlock() if err := store.loadPortClearsLocked(); err != nil { return nil, err } if err := store.loadLinkClearLocked(); err != nil { return nil, err } if err := store.loadSystemClearLocked(); err != nil { return nil, err } if err := store.rebuildLocked(); err != nil { return nil, err } if len(store.portClears.Cutoffs) > 0 { if err := store.compactClearedPortsLocked(); err != nil { return nil, err } } if store.systemClear.Revision > 0 { if err := store.compactClearedClassLocked(ClassSystem, store.systemRecordCleared); err != nil { return nil, err } if err := store.compactClearedClassLocked(ClassAudit, store.systemRecordCleared); err != nil { return nil, err } } if store.linkClear.Revision > 0 { if err := store.compactClearedClassLocked(ClassLink, store.linkRecordCleared); err != nil { return nil, err } } return store, nil } func ensureDirectory(path string, private bool) error { mode := fs.FileMode(0o750) if private { mode = 0o700 } info, err := os.Lstat(path) if os.IsNotExist(err) { if err := os.Mkdir(path, mode); err != nil { return fmt.Errorf("create log storage directory: %w", err) } return nil } if err != nil { return fmt.Errorf("inspect log storage directory: %w", err) } if !info.IsDir() || info.Mode()&os.ModeSymlink != 0 { return fmt.Errorf("log storage path must be a real directory: %s", path) } return nil } func (s *Store) Append(class string, timestamp time.Time, value any) (Envelope, error) { class, err := normalizeClass(class) if err != nil { return Envelope{}, err } if timestamp.IsZero() { timestamp = s.clock().UTC() } else { timestamp = timestamp.UTC() } payload, err := json.Marshal(value) if err != nil { return Envelope{}, fmt.Errorf("encode %s log payload: %w", class, err) } if len(payload) == 0 || int64(len(payload)) > maximumRecordBytes { return Envelope{}, fmt.Errorf("%s log payload must be between 1 and %d bytes", class, maximumRecordBytes) } s.mu.Lock() defer s.mu.Unlock() state := s.classes[class] if state == nil { state = &classState{NextSequence: 1} s.classes[class] = state } envelope := Envelope{FormatVersion: formatVersion, Sequence: state.NextSequence, Timestamp: timestamp, Class: class, Payload: payload} envelope.Checksum, err = envelopeChecksum(envelope) if err != nil { return Envelope{}, err } line, err := json.Marshal(envelope) if err != nil { return Envelope{}, fmt.Errorf("encode %s log envelope: %w", class, err) } line = append(line, '\n') if int64(len(line)) > maximumRecordBytes+4096 { return Envelope{}, fmt.Errorf("%s log record is too large", class) } if state.ActiveBytes > 0 && state.ActiveBytes+int64(len(line)) > s.segment { if err := s.rotateLocked(class); err != nil { return Envelope{}, err } } if err := s.ensureCapacityLocked(class, int64(len(line))); err != nil { return Envelope{}, err } path := s.activePath(class) file, err := openAppendRegular(path) if err != nil { return Envelope{}, fmt.Errorf("open active %s log segment: %w", class, err) } _, writeErr := file.Write(line) if writeErr == nil { writeErr = file.Sync() } closeErr := file.Close() if writeErr != nil { return Envelope{}, fmt.Errorf("append active %s log segment: %w", class, writeErr) } if closeErr != nil { return Envelope{}, fmt.Errorf("close active %s log segment: %w", class, closeErr) } if state.ActiveFirst.IsZero() { state.ActiveFirst = timestamp state.ActiveFirstSequence = envelope.Sequence } state.ActiveLast = timestamp state.ActiveLastSequence = envelope.Sequence if state.ActiveRecords%pageCheckpointRecords == 0 { state.ActiveOffsets = append(state.ActiveOffsets, recordOffset{Sequence: envelope.Sequence, Offset: state.ActiveBytes}) } state.ActiveRecords++ state.ActiveBytes += int64(len(line)) state.NextSequence++ return envelope, nil } func openAppendRegular(path string) (*os.File, error) { if info, err := os.Lstat(path); err == nil { if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 { return nil, errors.New("active log segment must be a regular file") } } else if !os.IsNotExist(err) { return nil, err } return os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600) } func (s *Store) Read(class string) ([]json.RawMessage, error) { class, err := normalizeClass(class) if err != nil { return nil, err } s.mu.Lock() defer s.mu.Unlock() var payloads []json.RawMessage err = s.walkClassLocked(class, false, func(envelope Envelope) error { payloads = append(payloads, append(json.RawMessage(nil), envelope.Payload...)) return nil }) if err != nil { return nil, err } if payloads == nil { payloads = []json.RawMessage{} } return payloads, nil } // Page returns verified authoritative records after the supplied sequence. // The bounded page contract is used by the local tera-logd socket so API // startup and queries never require an unbounded response frame. func (s *Store) Page(class string, afterSequence uint64, limit int) ([]Envelope, error) { class, err := normalizeClass(class) if err != nil { return nil, err } if limit < 1 || limit > 500 { return nil, errors.New("log page limit must be between 1 and 500") } s.mu.Lock() defer s.mu.Unlock() result := make([]Envelope, 0, limit) visit := func(envelope Envelope) error { if s.recordCleared(envelope) { return nil } if envelope.Sequence > afterSequence && len(result) < limit { envelope.Payload = append(json.RawMessage(nil), envelope.Payload...) result = append(result, envelope) if len(result) == limit { return errPageComplete } } return nil } // Closed segments are immutable and were fully verified while the store was // opened. Use their indexed sequence ranges and sparse byte checkpoints to // avoid rescanning every older record for each socket page. Every returned // envelope is still decoded and checksum-verified below. for _, meta := range s.closed[class] { if meta.LastSequence <= afterSequence { continue } path := filepath.Join(s.bucketPath(class), meta.Path) offset := pageOffsetAfter(meta.PageOffsets, afterSequence) if err := readSegmentPage(path, class, offset, visit); err != nil && !errors.Is(err, errPageComplete) { return nil, err } if len(result) == limit { return result, nil } } state := s.classes[class] if state == nil || state.ActiveLastSequence <= afterSequence { return result, nil } active := s.activePath(class) if _, err := os.Lstat(active); err == nil { offset := pageOffsetAfter(state.ActiveOffsets, afterSequence) if err := readSegmentPage(active, class, offset, visit); err != nil && !errors.Is(err, errPageComplete) { return nil, err } } else if !os.IsNotExist(err) { return nil, err } return result, nil } func pageOffsetAfter(offsets []recordOffset, afterSequence uint64) int64 { var offset int64 for _, checkpoint := range offsets { if checkpoint.Sequence > afterSequence { break } offset = checkpoint.Offset } return offset } func readSegmentPage(path, class string, offset int64, visit func(Envelope) error) error { info, err := os.Lstat(path) if err != nil { return err } if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 { return fmt.Errorf("log segment must be a regular file: %s", path) } file, err := os.Open(path) if err != nil { return fmt.Errorf("read %s log segment: %w", class, err) } defer file.Close() if offset > 0 { if _, err := file.Seek(offset, io.SeekStart); err != nil { return fmt.Errorf("seek %s log segment: %w", class, err) } } reader := bufio.NewReaderSize(file, int(maximumRecordBytes+4096)) for { line, readErr := reader.ReadBytes('\n') if len(line) == 0 && errors.Is(readErr, io.EOF) { return nil } if readErr != nil { return fmt.Errorf("%w: %s: incomplete log record", ErrCorruptSegment, filepath.Base(path)) } line = bytes.TrimSpace(line) if len(line) == 0 { continue } var envelope Envelope if err := json.Unmarshal(line, &envelope); err != nil { return fmt.Errorf("%w: %s: %v", ErrCorruptSegment, filepath.Base(path), err) } if err := validateEnvelope(envelope, class); err != nil { return fmt.Errorf("%w: %s: %v", ErrCorruptSegment, filepath.Base(path), err) } if err := visit(envelope); err != nil { return err } } } func (s *Store) Export(class string, writer io.Writer) error { class, err := normalizeClass(class) if err != nil { return err } if writer == nil { return errors.New("log export writer is required") } s.mu.Lock() defer s.mu.Unlock() encoder := json.NewEncoder(writer) return s.walkClassLocked(class, false, func(envelope Envelope) error { return encoder.Encode(envelope) }) } func (s *Store) ApplyRetention(retention Retention) error { if err := validateRetention(retention); err != nil { return err } s.mu.Lock() defer s.mu.Unlock() s.retention = retention for _, class := range []string{ClassAudit, ClassSystem, ClassPort, ClassLink} { if state := s.classes[class]; state != nil && state.ActiveBytes > 0 { if err := s.rotateLocked(class); err != nil { return err } } } now := s.clock().UTC() for _, class := range []string{ClassAudit, ClassSystem, ClassPort, ClassLink} { cutoff := now.Add(-time.Duration(s.retentionDays(class)) * 24 * time.Hour) segments, err := s.closedSegmentsLocked(class) if err != nil { return err } for _, segment := range segments { meta, err := inspectSegment(segment, class, false, nil) if err != nil { return err } if !meta.Newest.IsZero() && meta.Newest.Before(cutoff) { if err := os.Remove(segment); err != nil { return fmt.Errorf("remove expired %s log segment: %w", class, err) } } } } return s.rebuildLocked() } func validateRetention(retention Retention) error { for name, days := range map[string]int{"audit_days": retention.AuditDays, "system_days": retention.SystemDays, "port_days": retention.PortDays} { if days < 1 || days > 365 { return fmt.Errorf("%s must be between 1 and 365", name) } } return nil } func (s *Store) Snapshot() (Snapshot, error) { s.mu.Lock() defer s.mu.Unlock() free, err := s.freeBytes(s.root) if err != nil { return Snapshot{}, fmt.Errorf("measure log filesystem free bytes: %w", err) } snapshot := Snapshot{ Root: s.root, FreeBytes: free, GlobalFreeFloor: s.floor, UsageBytes: make(map[string]int64), NextSequence: make(map[string]uint64), CapBytes: make(map[string]int64), Oldest: make(map[string]time.Time), Newest: make(map[string]time.Time), } for _, bucket := range []string{"audit", "system", "port", "index", "export"} { usage, err := directoryUsage(filepath.Join(s.root, bucket)) if err != nil { return Snapshot{}, err } snapshot.UsageBytes[bucket] = usage if bucket == "index" { snapshot.CapBytes[bucket] = s.indexCap } else if capBytes, ok := s.caps[bucket]; ok { snapshot.CapBytes[bucket] = capBytes } } for _, class := range []string{ClassAudit, ClassSystem, ClassPort, ClassLink} { state := s.classes[class] if state != nil { snapshot.NextSequence[class] = state.NextSequence } for _, meta := range s.closed[class] { if snapshot.Oldest[class].IsZero() || meta.Oldest.Before(snapshot.Oldest[class]) { snapshot.Oldest[class] = meta.Oldest } if meta.Newest.After(snapshot.Newest[class]) { snapshot.Newest[class] = meta.Newest } } if state != nil && state.ActiveBytes > 0 { if snapshot.Oldest[class].IsZero() || state.ActiveFirst.Before(snapshot.Oldest[class]) { snapshot.Oldest[class] = state.ActiveFirst } if state.ActiveLast.After(snapshot.Newest[class]) { snapshot.Newest[class] = state.ActiveLast } } } return snapshot, nil } func (s *Store) RebuildIndex() error { s.mu.Lock() defer s.mu.Unlock() return s.rebuildLocked() } func (s *Store) rebuildLocked() error { s.classes = make(map[string]*classState) s.closed = make(map[string][]segmentMeta) for _, class := range []string{ClassAudit, ClassSystem, ClassPort, ClassLink} { state := &classState{NextSequence: 1} var lastSequence uint64 segments, err := s.closedSegmentsLocked(class) if err != nil { return err } for _, path := range segments { meta, err := inspectSegment(path, class, false, func(envelope Envelope) error { if envelope.Sequence <= lastSequence { return fmt.Errorf("%w: %s sequence is not increasing", ErrCorruptSegment, class) } lastSequence = envelope.Sequence return nil }) if err != nil { return err } s.closed[class] = append(s.closed[class], meta) } active := s.activePath(class) if info, err := os.Lstat(active); err == nil { if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 { return fmt.Errorf("active %s log segment must be a regular file", class) } meta, err := inspectSegment(active, class, true, func(envelope Envelope) error { if envelope.Sequence <= lastSequence { return fmt.Errorf("%w: %s sequence is not increasing", ErrCorruptSegment, class) } lastSequence = envelope.Sequence return nil }) if err != nil { return err } state.ActiveBytes, state.ActiveFirst, state.ActiveLast = meta.Bytes, meta.Oldest, meta.Newest state.ActiveFirstSequence, state.ActiveLastSequence = meta.FirstSequence, meta.LastSequence state.ActiveRecords = meta.Records state.ActiveOffsets = append([]recordOffset(nil), meta.PageOffsets...) } else if !os.IsNotExist(err) { return err } state.NextSequence = lastSequence + 1 if class == ClassPort && state.NextSequence < s.portClears.NextSequence { state.NextSequence = s.portClears.NextSequence } if class == ClassLink && state.NextSequence < s.linkClear.NextSequence { state.NextSequence = s.linkClear.NextSequence } if class == ClassAudit && state.NextSequence < s.systemClear.AuditNextSequence { state.NextSequence = s.systemClear.AuditNextSequence } if class == ClassSystem && state.NextSequence < s.systemClear.NextSequence { state.NextSequence = s.systemClear.NextSequence } s.classes[class] = state } return s.writeIndexLocked() } func (s *Store) walkClassLocked(class string, recoverActive bool, visit func(Envelope) error) error { if visit != nil { original := visit visit = func(event Envelope) error { if s.recordCleared(event) { return nil } return original(event) } } segments, err := s.closedSegmentsLocked(class) if err != nil { return err } for _, segment := range segments { if _, err := inspectSegment(segment, class, false, visit); err != nil { return err } } active := s.activePath(class) if _, err := os.Lstat(active); err == nil { _, err = inspectSegment(active, class, recoverActive, visit) return err } else if !os.IsNotExist(err) { return err } return nil } func inspectSegment(path, class string, recoverTail bool, visit func(Envelope) error) (segmentMeta, error) { info, err := os.Lstat(path) if err != nil { return segmentMeta{}, err } if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 { return segmentMeta{}, fmt.Errorf("log segment must be a regular file: %s", path) } data, err := os.ReadFile(path) if err != nil { return segmentMeta{}, fmt.Errorf("read %s log segment: %w", class, err) } meta := segmentMeta{Class: class, Path: filepath.Base(path), Closed: !recoverTail, Bytes: int64(len(data))} validBytes := 0 parts := bytes.SplitAfter(data, []byte{'\n'}) for index, part := range parts { if len(part) == 0 { continue } complete := part[len(part)-1] == '\n' line := bytes.TrimSpace(part) if len(line) == 0 { validBytes += len(part) continue } var envelope Envelope decodeErr := json.Unmarshal(line, &envelope) if decodeErr == nil { decodeErr = validateEnvelope(envelope, class) } isTail := index == len(parts)-1 || index == len(parts)-2 && len(parts[len(parts)-1]) == 0 if decodeErr != nil || !complete { if recoverTail && isTail { if err := os.Truncate(path, int64(validBytes)); err != nil { return segmentMeta{}, fmt.Errorf("recover truncated %s log tail: %w", class, err) } meta.Bytes = int64(validBytes) break } return segmentMeta{}, fmt.Errorf("%w: %s: %v", ErrCorruptSegment, filepath.Base(path), decodeErr) } if meta.FirstSequence == 0 { meta.FirstSequence, meta.Oldest = envelope.Sequence, envelope.Timestamp } meta.LastSequence, meta.Newest = envelope.Sequence, envelope.Timestamp if meta.Records%pageCheckpointRecords == 0 { meta.PageOffsets = append(meta.PageOffsets, recordOffset{Sequence: envelope.Sequence, Offset: int64(validBytes)}) } meta.Records++ if visit != nil { if err := visit(envelope); err != nil { return segmentMeta{}, err } } validBytes += len(part) } return meta, nil } func validateEnvelope(envelope Envelope, class string) error { if envelope.FormatVersion != formatVersion || envelope.Class != class || envelope.Sequence == 0 || envelope.Timestamp.IsZero() || len(envelope.Payload) == 0 { return errors.New("invalid log envelope identity") } expected, err := envelopeChecksum(envelope) if err != nil { return err } if !strings.EqualFold(expected, envelope.Checksum) { return errors.New("checksum mismatch") } return nil } func envelopeChecksum(envelope Envelope) (string, error) { core := struct { FormatVersion int `json:"format_version"` Sequence uint64 `json:"sequence"` Timestamp time.Time `json:"timestamp"` Class string `json:"class"` Payload json.RawMessage `json:"payload"` }{envelope.FormatVersion, envelope.Sequence, envelope.Timestamp, envelope.Class, envelope.Payload} data, err := json.Marshal(core) if err != nil { return "", err } sum := sha256.Sum256(data) return hex.EncodeToString(sum[:]), nil } func (s *Store) rotateLocked(class string) error { state := s.classes[class] if state == nil || state.ActiveBytes == 0 { return nil } active := s.activePath(class) meta, err := inspectSegment(active, class, true, nil) if err != nil { return err } if meta.FirstSequence == 0 { return nil } closedName := fmt.Sprintf("segment-%s-%020d-%020d.jsonl", class, meta.FirstSequence, meta.LastSequence) closed := filepath.Join(s.bucketPath(class), closedName) if _, err := os.Lstat(closed); err == nil { return fmt.Errorf("closed log segment already exists: %s", closedName) } else if !os.IsNotExist(err) { return err } if err := os.Rename(active, closed); err != nil { return fmt.Errorf("close %s log segment: %w", class, err) } if err := syncDirectory(s.bucketPath(class)); err != nil { return err } meta.Path = closedName meta.Closed = true s.closed[class] = append(s.closed[class], meta) state.ActiveBytes, state.ActiveFirst, state.ActiveLast = 0, time.Time{}, time.Time{} state.ActiveFirstSequence, state.ActiveLastSequence = 0, 0 state.ActiveRecords, state.ActiveOffsets = 0, nil return s.writeIndexLocked() } func (s *Store) ensureCapacityLocked(class string, incoming int64) error { bucket := bucketForClass(class) capBytes := s.caps[bucket] if capBytes > 0 { if incoming > capBytes { return fmt.Errorf("%w: incoming record exceeds class capacity", ErrCapacityExhausted) } usage, err := directoryUsage(filepath.Join(s.root, bucket)) if err != nil { return err } if usage+incoming > capBytes { if err := s.sealBucketLocked(bucket); err != nil { return err } if err := s.reclaimOldestLocked(bucket, usage+incoming-capBytes); err != nil { return err } usage, err = directoryUsage(filepath.Join(s.root, bucket)) if err != nil { return err } if usage+incoming > capBytes { return fmt.Errorf("%w: class=%s used=%d incoming=%d cap=%d", ErrCapacityExhausted, class, usage, incoming, capBytes) } } } if s.floor > 0 { free, err := s.freeBytes(s.root) if err != nil { return fmt.Errorf("measure log filesystem free bytes: %w", err) } if free-incoming < s.floor { needed := s.floor - (free - incoming) if err := s.reclaimGlobalOldestLocked(needed); err != nil { return err } free, err = s.freeBytes(s.root) if err != nil || free-incoming < s.floor { return fmt.Errorf("%w: free=%d incoming=%d floor=%d", ErrCapacityExhausted, free, incoming, s.floor) } } } return nil } func (s *Store) sealBucketLocked(bucket string) error { for _, class := range []string{ClassAudit, ClassSystem, ClassPort, ClassLink} { if bucketForClass(class) == bucket && s.classes[class] != nil && s.classes[class].ActiveBytes > 0 { if err := s.rotateLocked(class); err != nil { return err } } } return nil } func (s *Store) reclaimGlobalOldestLocked(needed int64) error { var candidates []segmentMeta for _, bucket := range []string{"audit", "system", "port"} { if err := s.sealBucketLocked(bucket); err != nil { return err } } for _, segments := range s.closed { candidates = append(candidates, segments...) } sort.Slice(candidates, func(i, j int) bool { if candidates[i].Oldest.Equal(candidates[j].Oldest) { return candidates[i].Path < candidates[j].Path } return candidates[i].Oldest.Before(candidates[j].Oldest) }) var reclaimed int64 for _, meta := range candidates { if err := os.Remove(filepath.Join(s.bucketPath(meta.Class), meta.Path)); err != nil { return err } s.removeClosedMetaLocked(meta.Class, meta.Path) reclaimed += meta.Bytes if reclaimed >= needed { return nil } } return fmt.Errorf("%w: insufficient log segments to rotate", ErrCapacityExhausted) } func (s *Store) reclaimOldestLocked(bucket string, needed int64) error { candidates, err := s.bucketSegmentsLocked(bucket) if err != nil { return err } var reclaimed int64 for _, path := range candidates { info, err := os.Lstat(path) if err != nil { return err } if err := os.Remove(path); err != nil { return err } reclaimed += info.Size() s.removeClosedMetaLocked(classFromSegment(path), filepath.Base(path)) if reclaimed >= needed { return nil } } return fmt.Errorf("%w: no reclaimable closed %s segments", ErrCapacityExhausted, bucket) } func (s *Store) bucketSegmentsLocked(bucket string) ([]string, error) { entries, err := os.ReadDir(filepath.Join(s.root, bucket)) if err != nil { return nil, err } paths := make([]string, 0) for _, entry := range entries { if entry.Type()&os.ModeSymlink != 0 { return nil, fmt.Errorf("symlink is not allowed in log storage: %s", entry.Name()) } if entry.Type().IsRegular() && strings.HasPrefix(entry.Name(), "segment-") && strings.HasSuffix(entry.Name(), ".jsonl") { paths = append(paths, filepath.Join(s.root, bucket, entry.Name())) } } sort.Slice(paths, func(i, j int) bool { iInfo, _ := os.Stat(paths[i]) jInfo, _ := os.Stat(paths[j]) if iInfo != nil && jInfo != nil && !iInfo.ModTime().Equal(jInfo.ModTime()) { return iInfo.ModTime().Before(jInfo.ModTime()) } return filepath.Base(paths[i]) < filepath.Base(paths[j]) }) return paths, nil } func (s *Store) closedSegmentsLocked(class string) ([]string, error) { entries, err := os.ReadDir(s.bucketPath(class)) if err != nil { return nil, err } prefix := "segment-" + class + "-" paths := make([]string, 0) for _, entry := range entries { if entry.Type()&os.ModeSymlink != 0 { return nil, fmt.Errorf("symlink is not allowed in log storage: %s", entry.Name()) } if entry.Type().IsRegular() && strings.HasPrefix(entry.Name(), prefix) && strings.HasSuffix(entry.Name(), ".jsonl") { paths = append(paths, filepath.Join(s.bucketPath(class), entry.Name())) } } sort.Strings(paths) return paths, nil } func (s *Store) writeIndexLocked() error { index := segmentIndex{FormatVersion: formatVersion, RebuiltAt: s.clock().UTC(), Classes: make(map[string]uint64)} for _, class := range []string{ClassAudit, ClassSystem, ClassPort, ClassLink} { if state := s.classes[class]; state != nil { index.Classes[class] = state.NextSequence } index.Segments = append(index.Segments, s.closed[class]...) if state := s.classes[class]; state != nil && state.ActiveBytes > 0 { index.Segments = append(index.Segments, segmentMeta{ Class: class, Path: filepath.Base(s.activePath(class)), Closed: false, FirstSequence: state.ActiveFirstSequence, LastSequence: state.ActiveLastSequence, Oldest: state.ActiveFirst, Newest: state.ActiveLast, Bytes: state.ActiveBytes, Records: state.ActiveRecords, PageOffsets: state.ActiveOffsets, }) } } data, err := json.MarshalIndent(index, "", " ") if err != nil { return err } if s.indexCap > 0 && int64(len(data)) > s.indexCap { return fmt.Errorf("%w: rebuilt index bytes=%d cap=%d", ErrCapacityExhausted, len(data), s.indexCap) } if err := atomicfile.Write(filepath.Join(s.root, "index", "log-index-v1.json"), append(data, '\n'), 0o600); err != nil { return fmt.Errorf("write rebuildable log index: %w", err) } return nil } func (s *Store) removeClosedMetaLocked(class, name string) { if class == "" { return } items := s.closed[class] filtered := items[:0] for _, item := range items { if item.Path != name { filtered = append(filtered, item) } } s.closed[class] = filtered } func (s *Store) activePath(class string) string { return filepath.Join(s.bucketPath(class), "active-"+class+".jsonl") } func (s *Store) bucketPath(class string) string { return filepath.Join(s.root, bucketForClass(class)) } func bucketForClass(class string) string { if class == ClassLink || class == ClassPort { return "port" } return class } func classFromSegment(path string) string { name := filepath.Base(path) for _, class := range []string{ClassAudit, ClassSystem, ClassPort, ClassLink} { if strings.HasPrefix(name, "segment-"+class+"-") { return class } } return "" } func normalizeClass(class string) (string, error) { class = strings.ToLower(strings.TrimSpace(class)) switch class { case ClassAudit, ClassSystem, ClassPort, ClassLink: return class, nil default: return "", errors.New("log class must be audit, system, port, or link") } } func (s *Store) retentionDays(class string) int { switch class { case ClassAudit: return s.retention.AuditDays case ClassSystem: return s.retention.SystemDays default: return s.retention.PortDays } } func directoryUsage(root string) (int64, error) { var total int64 err := filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error { if walkErr != nil { return walkErr } if entry.Type()&os.ModeSymlink != 0 { return fmt.Errorf("symlink is not allowed in log storage: %s", path) } if entry.Type().IsRegular() { info, err := entry.Info() if err != nil { return err } total += info.Size() } return nil }) return total, err }