package logstorage import ( "bytes" "encoding/json" "errors" "fmt" "os" "path/filepath" "regexp" "tera-clone/services/tera-api/internal/storage/atomicfile" ) var canonicalPort = regexp.MustCompile(`^CH(?:0[1-9]|[1-3][0-9]|4[0-8])$`) // A small, authoritative deletion checkpoint is committed before compaction. // A crash between segment rewrites cannot resurrect cleared records, or reuse // their sequence numbers. This is not part of the rebuildable index. type portClearState struct { Version int `json:"version"` Revision uint64 `json:"revision"` NextSequence uint64 `json:"next_sequence"` Cutoffs map[string]uint64 `json:"cutoffs"` } func (s *Store) portClearPath() string { return filepath.Join(s.root, "port", "cleared-ports-v1.json") } func (s *Store) loadPortClearsLocked() error { info, err := os.Lstat(s.portClearPath()) if os.IsNotExist(err) { return nil } if err != nil { return err } if !info.Mode().IsRegular() || info.Size() > 8192 { return errors.New("invalid port log deletion checkpoint") } data, err := os.ReadFile(s.portClearPath()) if err != nil { return err } var state portClearState if err := json.Unmarshal(data, &state); err != nil { return err } if state.Version != 1 || state.NextSequence == 0 || len(state.Cutoffs) > 48 { return errors.New("invalid port log deletion checkpoint") } for channel, cutoff := range state.Cutoffs { if !canonicalPort.MatchString(channel) || cutoff >= state.NextSequence { return errors.New("invalid port log deletion cutoff") } } s.portClears = state return nil } func (s *Store) portRecordCleared(event Envelope) bool { if event.Class != ClassPort || len(s.portClears.Cutoffs) == 0 { return false } var record struct { Channel string `json:"channel"` } if json.Unmarshal(event.Payload, &record) != nil { return false } cutoff, found := s.portClears.Cutoffs[record.Channel] return found && event.Sequence <= cutoff } func (s *Store) PortClearRevision() uint64 { s.mu.Lock() defer s.mu.Unlock() return s.portClears.Revision } // ClearPort removes one port or ALL ports' communication logs. Audit, // system, and link-state classes are never deletion targets. The mutex gives // the clear a precise boundary relative to concurrent appends. func (s *Store) ClearPort(channel string) (int, error) { if channel != "ALL" && !canonicalPort.MatchString(channel) { return 0, errors.New("port log channel must be ALL or CH01 through CH48") } s.mu.Lock() defer s.mu.Unlock() deleted := 0 if err := s.walkClassLocked(ClassPort, false, func(event Envelope) error { var record struct { Channel string `json:"channel"` } if err := json.Unmarshal(event.Payload, &record); err != nil { return fmt.Errorf("invalid port log payload: %w", err) } if channel == "ALL" || record.Channel == channel { deleted++ } return nil }); err != nil { return 0, err } next := portClearState{Version: 1, Revision: s.portClears.Revision + 1, NextSequence: s.classes[ClassPort].NextSequence, Cutoffs: map[string]uint64{}} for key, value := range s.portClears.Cutoffs { next.Cutoffs[key] = value } if channel == "ALL" { for i := 1; i <= 48; i++ { next.Cutoffs[fmt.Sprintf("CH%02d", i)] = next.NextSequence - 1 } } else { next.Cutoffs[channel] = next.NextSequence - 1 } data, err := json.Marshal(next) if err != nil { return 0, err } if err := atomicfile.Write(s.portClearPath(), append(data, '\n'), 0o600); err != nil { // A directory-sync failure can occur after rename. Reload the committed // checkpoint so subsequent reads still honor a completed rename. _ = s.loadPortClearsLocked() return 0, err } s.portClears = next if err := s.compactClearedPortsLocked(); err != nil { return deleted, err } return deleted, nil } func (s *Store) compactClearedPortsLocked() error { return s.compactClearedClassLocked(ClassPort, s.portRecordCleared) } func (s *Store) compactClearedClassLocked(class string, cleared func(Envelope) bool) (resultErr error) { // Even a partial compaction changes byte offsets. Never leave an old index // in memory after an I/O failure halfway through the segment list. defer func() { if err := s.rebuildLocked(); resultErr == nil { resultErr = err } }() paths, err := s.closedSegmentsLocked(class) if err != nil { return err } active := s.activePath(class) if _, err := os.Lstat(active); err == nil { paths = append(paths, active) } else if !os.IsNotExist(err) { return err } for _, path := range paths { var kept bytes.Buffer removed := 0 _, err := inspectSegment(path, class, false, func(event Envelope) error { if cleared(event) { removed++ return nil } return json.NewEncoder(&kept).Encode(event) }) if err != nil { return err } if removed == 0 { continue } // The checkpoint already hides all targets. Each bounded segment rewrite // is atomic; remaining ports retain their original envelopes/checksums. if err := atomicfile.Write(path, kept.Bytes(), 0o600); err != nil { return err } } return nil }