Something went wrong. Try again.
A Transparent and Verifiable Way to Sync the AT Protocol's PLC Directory
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412package bundle
import ( "bufio" "bytes" "encoding/json" "fmt" "os" "path/filepath" "sync" "time"
"tangled.org/atscan.net/plcbundle/plc")
const MEMPOOL_FILE_PREFIX = "plc_mempool_"
// Mempool stores operations waiting to be bundled// Operations must be strictly chronologicaltype Mempool struct { operations []plc.PLCOperation targetBundle int // Which bundle number these operations are for minTimestamp time.Time // Operations must be after this time file string mu sync.RWMutex logger Logger validated bool // Track if we've validated chronological order}
// NewMempool creates a new mempool for a specific bundle numberfunc NewMempool(bundleDir string, targetBundle int, minTimestamp time.Time, logger Logger) (*Mempool, error) { filename := fmt.Sprintf("%s%06d.jsonl", MEMPOOL_FILE_PREFIX, targetBundle)
m := &Mempool{ file: filepath.Join(bundleDir, filename), targetBundle: targetBundle, minTimestamp: minTimestamp, operations: make([]plc.PLCOperation, 0), logger: logger, validated: false, }
// Load existing mempool from disk if it exists if err := m.Load(); err != nil { // If file doesn't exist, that's OK if !os.IsNotExist(err) { return nil, fmt.Errorf("failed to load mempool: %w", err) } }
return m, nil}
// Add adds operations to the mempool with strict validationfunc (m *Mempool) Add(ops []plc.PLCOperation) (int, error) { m.mu.Lock() defer m.mu.Unlock()
if len(ops) == 0 { return 0, nil }
// Build existing CID set existingCIDs := make(map[string]bool) for _, op := range m.operations { existingCIDs[op.CID] = true }
// Validate and add operations var newOps []plc.PLCOperation var lastTime time.Time
// Start from last operation time if we have any if len(m.operations) > 0 { lastTime = m.operations[len(m.operations)-1].CreatedAt } else { lastTime = m.minTimestamp }
for _, op := range ops { // Skip duplicates if existingCIDs[op.CID] { continue }
// CRITICAL: Validate chronological order if !op.CreatedAt.After(lastTime) && !op.CreatedAt.Equal(lastTime) { return len(newOps), fmt.Errorf( "chronological violation: operation %s at %s is not after %s", op.CID, op.CreatedAt.Format(time.RFC3339Nano), lastTime.Format(time.RFC3339Nano), ) }
// Validate operation is after minimum timestamp if op.CreatedAt.Before(m.minTimestamp) { return len(newOps), fmt.Errorf( "operation %s at %s is before minimum timestamp %s (belongs in earlier bundle)", op.CID, op.CreatedAt.Format(time.RFC3339Nano), m.minTimestamp.Format(time.RFC3339Nano), ) }
newOps = append(newOps, op) existingCIDs[op.CID] = true lastTime = op.CreatedAt }
// Add new operations m.operations = append(m.operations, newOps...) m.validated = true
return len(newOps), nil}
// Validate performs a full chronological validation of all operationsfunc (m *Mempool) Validate() error { m.mu.RLock() defer m.mu.RUnlock()
if len(m.operations) == 0 { return nil }
// Check all operations are after minimum timestamp for i, op := range m.operations { if op.CreatedAt.Before(m.minTimestamp) { return fmt.Errorf( "operation %d (CID: %s) at %s is before minimum timestamp %s", i, op.CID, op.CreatedAt.Format(time.RFC3339Nano), m.minTimestamp.Format(time.RFC3339Nano), ) } }
// Check chronological order for i := 1; i < len(m.operations); i++ { prev := m.operations[i-1] curr := m.operations[i]
if curr.CreatedAt.Before(prev.CreatedAt) { return fmt.Errorf( "chronological violation at index %d: %s (%s) is before %s (%s)", i, curr.CID, curr.CreatedAt.Format(time.RFC3339Nano), prev.CID, prev.CreatedAt.Format(time.RFC3339Nano), ) } }
// Check for duplicate CIDs cidSet := make(map[string]int) for i, op := range m.operations { if prevIdx, exists := cidSet[op.CID]; exists { return fmt.Errorf( "duplicate CID %s at indices %d and %d", op.CID, prevIdx, i, ) } cidSet[op.CID] = i }
return nil}
// Count returns the number of operations in mempoolfunc (m *Mempool) Count() int { m.mu.RLock() defer m.mu.RUnlock() return len(m.operations)}
// Take removes and returns up to n operations from the frontfunc (m *Mempool) Take(n int) ([]plc.PLCOperation, error) { m.mu.Lock() defer m.mu.Unlock()
// Validate before taking if err := m.validateLocked(); err != nil { return nil, fmt.Errorf("mempool validation failed: %w", err) }
if n > len(m.operations) { n = len(m.operations) }
result := make([]plc.PLCOperation, n) copy(result, m.operations[:n])
// Remove taken operations m.operations = m.operations[n:]
return result, nil}
// validateLocked performs validation with lock already heldfunc (m *Mempool) validateLocked() error { if m.validated { return nil }
if len(m.operations) == 0 { return nil }
// Check chronological order lastTime := m.minTimestamp for i, op := range m.operations { if op.CreatedAt.Before(lastTime) { return fmt.Errorf( "chronological violation at index %d: %s is before %s", i, op.CreatedAt.Format(time.RFC3339Nano), lastTime.Format(time.RFC3339Nano), ) } lastTime = op.CreatedAt }
m.validated = true return nil}
// Peek returns up to n operations without removing themfunc (m *Mempool) Peek(n int) []plc.PLCOperation { m.mu.RLock() defer m.mu.RUnlock()
if n > len(m.operations) { n = len(m.operations) }
result := make([]plc.PLCOperation, n) copy(result, m.operations[:n])
return result}
// Clear removes all operationsfunc (m *Mempool) Clear() { m.mu.Lock() defer m.mu.Unlock() m.operations = make([]plc.PLCOperation, 0) m.validated = false}
// Save persists mempool to diskfunc (m *Mempool) Save() error { m.mu.RLock() defer m.mu.RUnlock()
if len(m.operations) == 0 { // Remove file if empty os.Remove(m.file) return nil }
// Validate before saving if err := m.validateLocked(); err != nil { return fmt.Errorf("mempool validation failed, refusing to save: %w", err) }
// Serialize to JSONL var buf bytes.Buffer for _, op := range m.operations { if len(op.RawJSON) > 0 { buf.Write(op.RawJSON) } else { data, _ := json.Marshal(op) buf.Write(data) } buf.WriteByte('\n') }
// Write atomically tempFile := m.file + ".tmp" if err := os.WriteFile(tempFile, buf.Bytes(), 0644); err != nil { return fmt.Errorf("failed to write mempool: %w", err) }
if err := os.Rename(tempFile, m.file); err != nil { os.Remove(tempFile) return fmt.Errorf("failed to rename mempool file: %w", err) }
return nil}
// Load reads mempool from disk and validates itfunc (m *Mempool) Load() error { data, err := os.ReadFile(m.file) if err != nil { return err }
m.mu.Lock() defer m.mu.Unlock()
// Parse JSONL scanner := bufio.NewScanner(bytes.NewReader(data)) buf := make([]byte, 0, 64*1024) scanner.Buffer(buf, 1024*1024)
m.operations = make([]plc.PLCOperation, 0)
for scanner.Scan() { line := scanner.Bytes() if len(line) == 0 { continue }
var op plc.PLCOperation if err := json.Unmarshal(line, &op); err != nil { return fmt.Errorf("failed to parse mempool operation: %w", err) }
op.RawJSON = make([]byte, len(line)) copy(op.RawJSON, line)
m.operations = append(m.operations, op) }
if err := scanner.Err(); err != nil { return fmt.Errorf("scanner error: %w", err) }
// CRITICAL: Validate loaded data if err := m.validateLocked(); err != nil { return fmt.Errorf("loaded mempool failed validation: %w", err) }
if len(m.operations) > 0 { m.logger.Printf("Loaded %d operations from mempool for bundle %06d", len(m.operations), m.targetBundle) }
return nil}
// GetFirstTime returns the created_at of the first operationfunc (m *Mempool) GetFirstTime() string { m.mu.RLock() defer m.mu.RUnlock()
if len(m.operations) == 0 { return "" }
return m.operations[0].CreatedAt.Format(time.RFC3339Nano)}
// GetLastTime returns the created_at of the last operationfunc (m *Mempool) GetLastTime() string { m.mu.RLock() defer m.mu.RUnlock()
if len(m.operations) == 0 { return "" }
return m.operations[len(m.operations)-1].CreatedAt.Format(time.RFC3339Nano)}
// GetTargetBundle returns the bundle number this mempool is forfunc (m *Mempool) GetTargetBundle() int { return m.targetBundle}
// GetMinTimestamp returns the minimum timestamp for operationsfunc (m *Mempool) GetMinTimestamp() time.Time { return m.minTimestamp}
// Stats returns mempool statisticsfunc (m *Mempool) Stats() map[string]interface{} { m.mu.RLock() defer m.mu.RUnlock()
count := len(m.operations)
stats := map[string]interface{}{ "count": count, "can_create_bundle": count >= BUNDLE_SIZE, "target_bundle": m.targetBundle, "min_timestamp": m.minTimestamp, "validated": m.validated, }
if count > 0 { stats["first_time"] = m.operations[0].CreatedAt stats["last_time"] = m.operations[len(m.operations)-1].CreatedAt
// Calculate size and unique DIDs totalSize := 0 didSet := make(map[string]bool) for _, op := range m.operations { totalSize += len(op.RawJSON) didSet[op.DID] = true } stats["size_bytes"] = totalSize stats["did_count"] = len(didSet) // ← ADDED }
return stats}
// Delete removes the mempool filefunc (m *Mempool) Delete() error { if err := os.Remove(m.file); err != nil && !os.IsNotExist(err) { return fmt.Errorf("failed to delete mempool file: %w", err) } return nil}
// GetFilename returns the mempool filenamefunc (m *Mempool) GetFilename() string { return filepath.Base(m.file)}