atscan.net
[DEPRECATED] Go implementation of plcbundle
update structure (3)
tree.fail
+19
-17
bundle/bundle_test.go
internal/bundle/bundle_test.go
+19
-17
bundle/bundle_test.go
internal/bundle/bundle_test.go
···
5
"testing"
6
"time"
7
8
-
"tangled.org/atscan.net/plcbundle/bundle"
9
"tangled.org/atscan.net/plcbundle/internal/storage"
10
"tangled.org/atscan.net/plcbundle/plcclient"
11
)
12
···
17
if idx == nil {
18
t.Fatal("NewIndex returned nil")
19
}
20
-
if idx.Version != bundle.INDEX_VERSION {
21
-
t.Errorf("expected version %s, got %s", bundle.INDEX_VERSION, idx.Version)
22
}
23
if idx.Count() != 0 {
24
t.Errorf("expected empty index, got count %d", idx.Count())
···
31
BundleNumber: 1,
32
StartTime: time.Now(),
33
EndTime: time.Now().Add(time.Hour),
34
-
OperationCount: bundle.BUNDLE_SIZE,
35
DIDCount: 1000,
36
Hash: "abc123",
37
CompressedHash: "def456",
···
62
BundleNumber: 1,
63
StartTime: time.Now(),
64
EndTime: time.Now().Add(time.Hour),
65
-
OperationCount: bundle.BUNDLE_SIZE,
66
Hash: "test123",
67
})
68
···
88
BundleNumber: i,
89
StartTime: time.Now(),
90
EndTime: time.Now().Add(time.Hour),
91
-
OperationCount: bundle.BUNDLE_SIZE,
92
})
93
}
94
···
109
BundleNumber: num,
110
StartTime: time.Now(),
111
EndTime: time.Now().Add(time.Hour),
112
-
OperationCount: bundle.BUNDLE_SIZE,
113
})
114
}
115
···
137
BundleNumber: 1,
138
StartTime: time.Now(),
139
EndTime: time.Now().Add(time.Hour),
140
-
Operations: makeTestOperations(bundle.BUNDLE_SIZE),
141
},
142
wantErr: false,
143
},
···
145
name: "invalid bundle number",
146
bundle: &bundle.Bundle{
147
BundleNumber: 0,
148
-
Operations: makeTestOperations(bundle.BUNDLE_SIZE),
149
},
150
wantErr: true,
151
},
···
163
BundleNumber: 1,
164
StartTime: time.Now().Add(time.Hour),
165
EndTime: time.Now(),
166
-
Operations: makeTestOperations(bundle.BUNDLE_SIZE),
167
},
168
wantErr: true,
169
},
···
198
199
t.Run("CreateAndAdd", func(t *testing.T) {
200
minTime := time.Now().Add(-time.Hour)
201
-
m, err := bundle.NewMempool(tmpDir, 1, minTime, logger)
202
if err != nil {
203
t.Fatalf("NewMempool failed: %v", err)
204
}
···
218
219
t.Run("ChronologicalValidation", func(t *testing.T) {
220
minTime := time.Now().Add(-time.Hour)
221
-
m, err := bundle.NewMempool(tmpDir, 2, minTime, logger)
222
if err != nil {
223
t.Fatalf("NewMempool failed: %v", err)
224
}
···
246
247
t.Run("TakeOperations", func(t *testing.T) {
248
minTime := time.Now().Add(-time.Hour)
249
-
m, err := bundle.NewMempool(tmpDir, 3, minTime, logger)
250
if err != nil {
251
t.Fatalf("NewMempool failed: %v", err)
252
}
···
268
269
t.Run("SaveAndLoad", func(t *testing.T) {
270
minTime := time.Now().Add(-time.Hour)
271
-
m, err := bundle.NewMempool(tmpDir, 4, minTime, logger)
272
if err != nil {
273
t.Fatalf("NewMempool failed: %v", err)
274
}
···
281
}
282
283
// Create new mempool and load
284
-
m2, err := bundle.NewMempool(tmpDir, 4, minTime, logger)
285
if err != nil {
286
t.Fatalf("NewMempool failed: %v", err)
287
}
···
293
294
t.Run("Validate", func(t *testing.T) {
295
minTime := time.Now().Add(-time.Hour)
296
-
m, err := bundle.NewMempool(tmpDir, 5, minTime, logger)
297
if err != nil {
298
t.Fatalf("NewMempool failed: %v", err)
299
}
···
341
})
342
343
t.Run("SaveAndLoadBundle", func(t *testing.T) {
344
-
operations := makeTestOperations(bundle.BUNDLE_SIZE)
345
path := filepath.Join(tmpDir, "test_bundle.jsonl.zst")
346
347
// Save
···
5
"testing"
6
"time"
7
8
+
"tangled.org/atscan.net/plcbundle/internal/bundle"
9
+
"tangled.org/atscan.net/plcbundle/internal/mempool"
10
"tangled.org/atscan.net/plcbundle/internal/storage"
11
+
"tangled.org/atscan.net/plcbundle/internal/types"
12
"tangled.org/atscan.net/plcbundle/plcclient"
13
)
14
···
19
if idx == nil {
20
t.Fatal("NewIndex returned nil")
21
}
22
+
if idx.Version != types.INDEX_VERSION {
23
+
t.Errorf("expected version %s, got %s", types.INDEX_VERSION, idx.Version)
24
}
25
if idx.Count() != 0 {
26
t.Errorf("expected empty index, got count %d", idx.Count())
···
33
BundleNumber: 1,
34
StartTime: time.Now(),
35
EndTime: time.Now().Add(time.Hour),
36
+
OperationCount: types.BUNDLE_SIZE,
37
DIDCount: 1000,
38
Hash: "abc123",
39
CompressedHash: "def456",
···
64
BundleNumber: 1,
65
StartTime: time.Now(),
66
EndTime: time.Now().Add(time.Hour),
67
+
OperationCount: types.BUNDLE_SIZE,
68
Hash: "test123",
69
})
70
···
90
BundleNumber: i,
91
StartTime: time.Now(),
92
EndTime: time.Now().Add(time.Hour),
93
+
OperationCount: types.BUNDLE_SIZE,
94
})
95
}
96
···
111
BundleNumber: num,
112
StartTime: time.Now(),
113
EndTime: time.Now().Add(time.Hour),
114
+
OperationCount: types.BUNDLE_SIZE,
115
})
116
}
117
···
139
BundleNumber: 1,
140
StartTime: time.Now(),
141
EndTime: time.Now().Add(time.Hour),
142
+
Operations: makeTestOperations(types.BUNDLE_SIZE),
143
},
144
wantErr: false,
145
},
···
147
name: "invalid bundle number",
148
bundle: &bundle.Bundle{
149
BundleNumber: 0,
150
+
Operations: makeTestOperations(types.BUNDLE_SIZE),
151
},
152
wantErr: true,
153
},
···
165
BundleNumber: 1,
166
StartTime: time.Now().Add(time.Hour),
167
EndTime: time.Now(),
168
+
Operations: makeTestOperations(types.BUNDLE_SIZE),
169
},
170
wantErr: true,
171
},
···
200
201
t.Run("CreateAndAdd", func(t *testing.T) {
202
minTime := time.Now().Add(-time.Hour)
203
+
m, err := mempool.NewMempool(tmpDir, 1, minTime, logger)
204
if err != nil {
205
t.Fatalf("NewMempool failed: %v", err)
206
}
···
220
221
t.Run("ChronologicalValidation", func(t *testing.T) {
222
minTime := time.Now().Add(-time.Hour)
223
+
m, err := mempool.NewMempool(tmpDir, 2, minTime, logger)
224
if err != nil {
225
t.Fatalf("NewMempool failed: %v", err)
226
}
···
248
249
t.Run("TakeOperations", func(t *testing.T) {
250
minTime := time.Now().Add(-time.Hour)
251
+
m, err := mempool.NewMempool(tmpDir, 3, minTime, logger)
252
if err != nil {
253
t.Fatalf("NewMempool failed: %v", err)
254
}
···
270
271
t.Run("SaveAndLoad", func(t *testing.T) {
272
minTime := time.Now().Add(-time.Hour)
273
+
m, err := mempool.NewMempool(tmpDir, 4, minTime, logger)
274
if err != nil {
275
t.Fatalf("NewMempool failed: %v", err)
276
}
···
283
}
284
285
// Create new mempool and load
286
+
m2, err := mempool.NewMempool(tmpDir, 4, minTime, logger)
287
if err != nil {
288
t.Fatalf("NewMempool failed: %v", err)
289
}
···
295
296
t.Run("Validate", func(t *testing.T) {
297
minTime := time.Now().Add(-time.Hour)
298
+
m, err := mempool.NewMempool(tmpDir, 5, minTime, logger)
299
if err != nil {
300
t.Fatalf("NewMempool failed: %v", err)
301
}
···
343
})
344
345
t.Run("SaveAndLoadBundle", func(t *testing.T) {
346
+
operations := makeTestOperations(types.BUNDLE_SIZE)
347
path := filepath.Join(tmpDir, "test_bundle.jsonl.zst")
348
349
// Save
+3
-2
bundle/bundler.go
internal/bundle/bundler.go
+3
-2
bundle/bundler.go
internal/bundle/bundler.go
···
3
import (
4
"time"
5
6
"tangled.org/atscan.net/plcbundle/plcclient"
7
)
8
9
// CreateBundle creates a complete bundle structure from operations
10
func (m *Manager) CreateBundle(bundleNumber int, operations []plcclient.PLCOperation, cursor string, parent string) *Bundle {
11
-
if len(operations) != BUNDLE_SIZE {
12
-
m.logger.Printf("Warning: bundle has %d operations, expected %d", len(operations), BUNDLE_SIZE)
13
}
14
15
dids := m.operations.ExtractUniqueDIDs(operations)
···
3
import (
4
"time"
5
6
+
"tangled.org/atscan.net/plcbundle/internal/types"
7
"tangled.org/atscan.net/plcbundle/plcclient"
8
)
9
10
// CreateBundle creates a complete bundle structure from operations
11
func (m *Manager) CreateBundle(bundleNumber int, operations []plcclient.PLCOperation, cursor string, parent string) *Bundle {
12
+
if len(operations) != types.BUNDLE_SIZE {
13
+
m.logger.Printf("Warning: bundle has %d operations, expected %d", len(operations), types.BUNDLE_SIZE)
14
}
15
16
dids := m.operations.ExtractUniqueDIDs(operations)
bundle/clone.go
internal/bundle/clone.go
bundle/clone.go
internal/bundle/clone.go
bundle/index.go
internal/bundle/index.go
bundle/index.go
internal/bundle/index.go
+18
-16
bundle/manager.go
internal/bundle/manager.go
+18
-16
bundle/manager.go
internal/bundle/manager.go
···
15
"time"
16
17
"tangled.org/atscan.net/plcbundle/internal/didindex"
18
"tangled.org/atscan.net/plcbundle/internal/storage"
19
"tangled.org/atscan.net/plcbundle/plcclient"
20
)
21
···
37
index *Index
38
indexPath string
39
plcClient *plcclient.Client
40
-
logger Logger
41
-
mempool *Mempool
42
didIndex *didindex.Manager // Updated type
43
44
bundleCache map[int]*Bundle
···
261
minTimestamp = lastBundle.EndTime
262
}
263
264
-
mempool, err := NewMempool(config.BundleDir, nextBundleNum, minTimestamp, config.Logger)
265
if err != nil {
266
return nil, fmt.Errorf("failed to initialize mempool: %w", err)
267
}
···
450
nextBundle := bundle.BundleNumber + 1
451
minTimestamp := bundle.EndTime
452
453
-
newMempool, err := NewMempool(m.config.BundleDir, nextBundle, minTimestamp, m.logger)
454
if err != nil {
455
return fmt.Errorf("failed to create new mempool: %w", err)
456
}
···
494
m.logger.Printf("Preparing bundle %06d (mempool: %d ops)...", nextBundleNum, m.mempool.Count())
495
}
496
497
-
for m.mempool.Count() < BUNDLE_SIZE {
498
if !quiet {
499
-
m.logger.Printf("Fetching more operations (have %d/%d)...", m.mempool.Count(), BUNDLE_SIZE)
500
}
501
502
-
err := m.fetchToMempool(ctx, afterTime, prevBoundaryCIDs, BUNDLE_SIZE-m.mempool.Count(), quiet)
503
if err != nil {
504
-
if m.mempool.Count() >= BUNDLE_SIZE {
505
break
506
}
507
m.mempool.Save()
508
-
return nil, fmt.Errorf("insufficient operations: have %d, need %d", m.mempool.Count(), BUNDLE_SIZE)
509
}
510
511
-
if m.mempool.Count() < BUNDLE_SIZE {
512
m.mempool.Save()
513
-
return nil, fmt.Errorf("insufficient operations: have %d, need %d (no more available)", m.mempool.Count(), BUNDLE_SIZE)
514
}
515
}
516
517
if !quiet {
518
m.logger.Printf("Creating bundle %06d from mempool", nextBundleNum)
519
}
520
-
operations, err := m.mempool.Take(BUNDLE_SIZE)
521
if err != nil {
522
m.mempool.Save()
523
return nil, fmt.Errorf("failed to take operations from mempool: %w", err)
···
1285
}
1286
1287
// GetMempool returns the current mempool
1288
-
func (m *Manager) GetMempool() *Mempool {
1289
return m.mempool
1290
}
1291
···
1309
func (m *Manager) GetCurrentCursor() int {
1310
index := m.GetIndex()
1311
bundles := index.GetBundles()
1312
-
cursor := len(bundles) * BUNDLE_SIZE
1313
1314
// Add mempool operations
1315
mempoolStats := m.GetMempoolStats()
···
1329
}
1330
1331
// Validate position
1332
-
if position < 0 || position >= BUNDLE_SIZE {
1333
-
return nil, fmt.Errorf("invalid position: %d (must be 0-%d)", position, BUNDLE_SIZE-1)
1334
}
1335
1336
// Build file path
···
15
"time"
16
17
"tangled.org/atscan.net/plcbundle/internal/didindex"
18
+
"tangled.org/atscan.net/plcbundle/internal/mempool"
19
"tangled.org/atscan.net/plcbundle/internal/storage"
20
+
"tangled.org/atscan.net/plcbundle/internal/types"
21
"tangled.org/atscan.net/plcbundle/plcclient"
22
)
23
···
39
index *Index
40
indexPath string
41
plcClient *plcclient.Client
42
+
logger types.Logger
43
+
mempool *mempool.Mempool
44
didIndex *didindex.Manager // Updated type
45
46
bundleCache map[int]*Bundle
···
263
minTimestamp = lastBundle.EndTime
264
}
265
266
+
mempool, err := mempool.NewMempool(config.BundleDir, nextBundleNum, minTimestamp, config.Logger)
267
if err != nil {
268
return nil, fmt.Errorf("failed to initialize mempool: %w", err)
269
}
···
452
nextBundle := bundle.BundleNumber + 1
453
minTimestamp := bundle.EndTime
454
455
+
newMempool, err := mempool.NewMempool(m.config.BundleDir, nextBundle, minTimestamp, m.logger)
456
if err != nil {
457
return fmt.Errorf("failed to create new mempool: %w", err)
458
}
···
496
m.logger.Printf("Preparing bundle %06d (mempool: %d ops)...", nextBundleNum, m.mempool.Count())
497
}
498
499
+
for m.mempool.Count() < types.BUNDLE_SIZE {
500
if !quiet {
501
+
m.logger.Printf("Fetching more operations (have %d/%d)...", m.mempool.Count(), types.BUNDLE_SIZE)
502
}
503
504
+
err := m.fetchToMempool(ctx, afterTime, prevBoundaryCIDs, types.BUNDLE_SIZE-m.mempool.Count(), quiet)
505
if err != nil {
506
+
if m.mempool.Count() >= types.BUNDLE_SIZE {
507
break
508
}
509
m.mempool.Save()
510
+
return nil, fmt.Errorf("insufficient operations: have %d, need %d", m.mempool.Count(), types.BUNDLE_SIZE)
511
}
512
513
+
if m.mempool.Count() < types.BUNDLE_SIZE {
514
m.mempool.Save()
515
+
return nil, fmt.Errorf("insufficient operations: have %d, need %d (no more available)", m.mempool.Count(), types.BUNDLE_SIZE)
516
}
517
}
518
519
if !quiet {
520
m.logger.Printf("Creating bundle %06d from mempool", nextBundleNum)
521
}
522
+
operations, err := m.mempool.Take(types.BUNDLE_SIZE)
523
if err != nil {
524
m.mempool.Save()
525
return nil, fmt.Errorf("failed to take operations from mempool: %w", err)
···
1287
}
1288
1289
// GetMempool returns the current mempool
1290
+
func (m *Manager) GetMempool() *mempool.Mempool {
1291
return m.mempool
1292
}
1293
···
1311
func (m *Manager) GetCurrentCursor() int {
1312
index := m.GetIndex()
1313
bundles := index.GetBundles()
1314
+
cursor := len(bundles) * types.BUNDLE_SIZE
1315
1316
// Add mempool operations
1317
mempoolStats := m.GetMempoolStats()
···
1331
}
1332
1333
// Validate position
1334
+
if position < 0 || position >= types.BUNDLE_SIZE {
1335
+
return nil, fmt.Errorf("invalid position: %d (must be 0-%d)", position, types.BUNDLE_SIZE-1)
1336
}
1337
1338
// Build file path
-418
bundle/mempool.go
-418
bundle/mempool.go
···
1
-
package bundle
2
-
3
-
import (
4
-
"bufio"
5
-
"bytes"
6
-
"fmt"
7
-
"os"
8
-
"path/filepath"
9
-
"sync"
10
-
"time"
11
-
12
-
"github.com/goccy/go-json"
13
-
"tangled.org/atscan.net/plcbundle/plcclient"
14
-
)
15
-
16
-
const MEMPOOL_FILE_PREFIX = "plc_mempool_"
17
-
18
-
// Mempool stores operations waiting to be bundled
19
-
// Operations must be strictly chronological
20
-
type Mempool struct {
21
-
operations []plcclient.PLCOperation
22
-
targetBundle int // Which bundle number these operations are for
23
-
minTimestamp time.Time // Operations must be after this time
24
-
file string
25
-
mu sync.RWMutex
26
-
logger Logger
27
-
validated bool // Track if we've validated chronological order
28
-
dirty bool // Track if mempool changed
29
-
}
30
-
31
-
// NewMempool creates a new mempool for a specific bundle number
32
-
func NewMempool(bundleDir string, targetBundle int, minTimestamp time.Time, logger Logger) (*Mempool, error) {
33
-
filename := fmt.Sprintf("%s%06d.jsonl", MEMPOOL_FILE_PREFIX, targetBundle)
34
-
35
-
m := &Mempool{
36
-
file: filepath.Join(bundleDir, filename),
37
-
targetBundle: targetBundle,
38
-
minTimestamp: minTimestamp,
39
-
operations: make([]plcclient.PLCOperation, 0),
40
-
logger: logger,
41
-
validated: false,
42
-
}
43
-
44
-
// Load existing mempool from disk if it exists
45
-
if err := m.Load(); err != nil {
46
-
// If file doesn't exist, that's OK
47
-
if !os.IsNotExist(err) {
48
-
return nil, fmt.Errorf("failed to load mempool: %w", err)
49
-
}
50
-
}
51
-
52
-
return m, nil
53
-
}
54
-
55
-
// Add adds operations to the mempool with strict validation
56
-
func (m *Mempool) Add(ops []plcclient.PLCOperation) (int, error) {
57
-
m.mu.Lock()
58
-
defer m.mu.Unlock()
59
-
60
-
if len(ops) == 0 {
61
-
return 0, nil
62
-
}
63
-
64
-
// Build existing CID set
65
-
existingCIDs := make(map[string]bool)
66
-
for _, op := range m.operations {
67
-
existingCIDs[op.CID] = true
68
-
}
69
-
70
-
// Validate and add operations
71
-
var newOps []plcclient.PLCOperation
72
-
var lastTime time.Time
73
-
74
-
// Start from last operation time if we have any
75
-
if len(m.operations) > 0 {
76
-
lastTime = m.operations[len(m.operations)-1].CreatedAt
77
-
} else {
78
-
lastTime = m.minTimestamp
79
-
}
80
-
81
-
for _, op := range ops {
82
-
// Skip duplicates
83
-
if existingCIDs[op.CID] {
84
-
continue
85
-
}
86
-
87
-
// CRITICAL: Validate chronological order
88
-
if !op.CreatedAt.After(lastTime) && !op.CreatedAt.Equal(lastTime) {
89
-
return len(newOps), fmt.Errorf(
90
-
"chronological violation: operation %s at %s is not after %s",
91
-
op.CID, op.CreatedAt.Format(time.RFC3339Nano), lastTime.Format(time.RFC3339Nano),
92
-
)
93
-
}
94
-
95
-
// Validate operation is after minimum timestamp
96
-
if op.CreatedAt.Before(m.minTimestamp) {
97
-
return len(newOps), fmt.Errorf(
98
-
"operation %s at %s is before minimum timestamp %s (belongs in earlier bundle)",
99
-
op.CID, op.CreatedAt.Format(time.RFC3339Nano), m.minTimestamp.Format(time.RFC3339Nano),
100
-
)
101
-
}
102
-
103
-
newOps = append(newOps, op)
104
-
existingCIDs[op.CID] = true
105
-
lastTime = op.CreatedAt
106
-
}
107
-
108
-
// Add new operations
109
-
m.operations = append(m.operations, newOps...)
110
-
m.validated = true
111
-
m.dirty = true
112
-
113
-
return len(newOps), nil
114
-
}
115
-
116
-
// Validate performs a full chronological validation of all operations
117
-
func (m *Mempool) Validate() error {
118
-
m.mu.RLock()
119
-
defer m.mu.RUnlock()
120
-
121
-
if len(m.operations) == 0 {
122
-
return nil
123
-
}
124
-
125
-
// Check all operations are after minimum timestamp
126
-
for i, op := range m.operations {
127
-
if op.CreatedAt.Before(m.minTimestamp) {
128
-
return fmt.Errorf(
129
-
"operation %d (CID: %s) at %s is before minimum timestamp %s",
130
-
i, op.CID, op.CreatedAt.Format(time.RFC3339Nano), m.minTimestamp.Format(time.RFC3339Nano),
131
-
)
132
-
}
133
-
}
134
-
135
-
// Check chronological order
136
-
for i := 1; i < len(m.operations); i++ {
137
-
prev := m.operations[i-1]
138
-
curr := m.operations[i]
139
-
140
-
if curr.CreatedAt.Before(prev.CreatedAt) {
141
-
return fmt.Errorf(
142
-
"chronological violation at index %d: %s (%s) is before %s (%s)",
143
-
i, curr.CID, curr.CreatedAt.Format(time.RFC3339Nano),
144
-
prev.CID, prev.CreatedAt.Format(time.RFC3339Nano),
145
-
)
146
-
}
147
-
}
148
-
149
-
// Check for duplicate CIDs
150
-
cidSet := make(map[string]int)
151
-
for i, op := range m.operations {
152
-
if prevIdx, exists := cidSet[op.CID]; exists {
153
-
return fmt.Errorf(
154
-
"duplicate CID %s at indices %d and %d",
155
-
op.CID, prevIdx, i,
156
-
)
157
-
}
158
-
cidSet[op.CID] = i
159
-
}
160
-
161
-
return nil
162
-
}
163
-
164
-
// Count returns the number of operations in mempool
165
-
func (m *Mempool) Count() int {
166
-
m.mu.RLock()
167
-
defer m.mu.RUnlock()
168
-
return len(m.operations)
169
-
}
170
-
171
-
// Take removes and returns up to n operations from the front
172
-
func (m *Mempool) Take(n int) ([]plcclient.PLCOperation, error) {
173
-
m.mu.Lock()
174
-
defer m.mu.Unlock()
175
-
176
-
// Validate before taking
177
-
if err := m.validateLocked(); err != nil {
178
-
return nil, fmt.Errorf("mempool validation failed: %w", err)
179
-
}
180
-
181
-
if n > len(m.operations) {
182
-
n = len(m.operations)
183
-
}
184
-
185
-
result := make([]plcclient.PLCOperation, n)
186
-
copy(result, m.operations[:n])
187
-
188
-
// Remove taken operations
189
-
m.operations = m.operations[n:]
190
-
191
-
return result, nil
192
-
}
193
-
194
-
// validateLocked performs validation with lock already held
195
-
func (m *Mempool) validateLocked() error {
196
-
if m.validated {
197
-
return nil
198
-
}
199
-
200
-
if len(m.operations) == 0 {
201
-
return nil
202
-
}
203
-
204
-
// Check chronological order
205
-
lastTime := m.minTimestamp
206
-
for i, op := range m.operations {
207
-
if op.CreatedAt.Before(lastTime) {
208
-
return fmt.Errorf(
209
-
"chronological violation at index %d: %s is before %s",
210
-
i, op.CreatedAt.Format(time.RFC3339Nano), lastTime.Format(time.RFC3339Nano),
211
-
)
212
-
}
213
-
lastTime = op.CreatedAt
214
-
}
215
-
216
-
m.validated = true
217
-
return nil
218
-
}
219
-
220
-
// Peek returns up to n operations without removing them
221
-
func (m *Mempool) Peek(n int) []plcclient.PLCOperation {
222
-
m.mu.RLock()
223
-
defer m.mu.RUnlock()
224
-
225
-
if n > len(m.operations) {
226
-
n = len(m.operations)
227
-
}
228
-
229
-
result := make([]plcclient.PLCOperation, n)
230
-
copy(result, m.operations[:n])
231
-
232
-
return result
233
-
}
234
-
235
-
// Clear removes all operations
236
-
func (m *Mempool) Clear() {
237
-
m.mu.Lock()
238
-
defer m.mu.Unlock()
239
-
m.operations = make([]plcclient.PLCOperation, 0)
240
-
m.validated = false
241
-
}
242
-
243
-
// Save persists mempool to disk
244
-
func (m *Mempool) Save() error {
245
-
m.mu.Lock()
246
-
defer m.mu.Unlock()
247
-
248
-
if !m.dirty {
249
-
return nil
250
-
}
251
-
252
-
if len(m.operations) == 0 {
253
-
// Remove file if empty
254
-
os.Remove(m.file)
255
-
return nil
256
-
}
257
-
258
-
// Validate before saving
259
-
if err := m.validateLocked(); err != nil {
260
-
return fmt.Errorf("mempool validation failed, refusing to save: %w", err)
261
-
}
262
-
263
-
// Serialize to JSONL
264
-
var buf bytes.Buffer
265
-
for _, op := range m.operations {
266
-
if len(op.RawJSON) > 0 {
267
-
buf.Write(op.RawJSON)
268
-
} else {
269
-
data, _ := json.Marshal(op)
270
-
buf.Write(data)
271
-
}
272
-
buf.WriteByte('\n')
273
-
}
274
-
275
-
// Write atomically
276
-
tempFile := m.file + ".tmp"
277
-
if err := os.WriteFile(tempFile, buf.Bytes(), 0644); err != nil {
278
-
return fmt.Errorf("failed to write mempool: %w", err)
279
-
}
280
-
281
-
if err := os.Rename(tempFile, m.file); err != nil {
282
-
os.Remove(tempFile)
283
-
return fmt.Errorf("failed to rename mempool file: %w", err)
284
-
}
285
-
286
-
m.dirty = false
287
-
return nil
288
-
}
289
-
290
-
// Load reads mempool from disk and validates it
291
-
func (m *Mempool) Load() error {
292
-
data, err := os.ReadFile(m.file)
293
-
if err != nil {
294
-
return err
295
-
}
296
-
297
-
m.mu.Lock()
298
-
defer m.mu.Unlock()
299
-
300
-
// Parse JSONL
301
-
scanner := bufio.NewScanner(bytes.NewReader(data))
302
-
buf := make([]byte, 0, 64*1024)
303
-
scanner.Buffer(buf, 1024*1024)
304
-
305
-
m.operations = make([]plcclient.PLCOperation, 0)
306
-
307
-
for scanner.Scan() {
308
-
line := scanner.Bytes()
309
-
if len(line) == 0 {
310
-
continue
311
-
}
312
-
313
-
var op plcclient.PLCOperation
314
-
if err := json.Unmarshal(line, &op); err != nil {
315
-
return fmt.Errorf("failed to parse mempool operation: %w", err)
316
-
}
317
-
318
-
op.RawJSON = make([]byte, len(line))
319
-
copy(op.RawJSON, line)
320
-
321
-
m.operations = append(m.operations, op)
322
-
}
323
-
324
-
if err := scanner.Err(); err != nil {
325
-
return fmt.Errorf("scanner error: %w", err)
326
-
}
327
-
328
-
// CRITICAL: Validate loaded data
329
-
if err := m.validateLocked(); err != nil {
330
-
return fmt.Errorf("loaded mempool failed validation: %w", err)
331
-
}
332
-
333
-
if len(m.operations) > 0 {
334
-
m.logger.Printf("Loaded %d operations from mempool for bundle %06d", len(m.operations), m.targetBundle)
335
-
}
336
-
337
-
return nil
338
-
}
339
-
340
-
// GetFirstTime returns the created_at of the first operation
341
-
func (m *Mempool) GetFirstTime() string {
342
-
m.mu.RLock()
343
-
defer m.mu.RUnlock()
344
-
345
-
if len(m.operations) == 0 {
346
-
return ""
347
-
}
348
-
349
-
return m.operations[0].CreatedAt.Format(time.RFC3339Nano)
350
-
}
351
-
352
-
// GetLastTime returns the created_at of the last operation
353
-
func (m *Mempool) GetLastTime() string {
354
-
m.mu.RLock()
355
-
defer m.mu.RUnlock()
356
-
357
-
if len(m.operations) == 0 {
358
-
return ""
359
-
}
360
-
361
-
return m.operations[len(m.operations)-1].CreatedAt.Format(time.RFC3339Nano)
362
-
}
363
-
364
-
// GetTargetBundle returns the bundle number this mempool is for
365
-
func (m *Mempool) GetTargetBundle() int {
366
-
return m.targetBundle
367
-
}
368
-
369
-
// GetMinTimestamp returns the minimum timestamp for operations
370
-
func (m *Mempool) GetMinTimestamp() time.Time {
371
-
return m.minTimestamp
372
-
}
373
-
374
-
// Stats returns mempool statistics
375
-
func (m *Mempool) Stats() map[string]interface{} {
376
-
m.mu.RLock()
377
-
defer m.mu.RUnlock()
378
-
379
-
count := len(m.operations)
380
-
381
-
stats := map[string]interface{}{
382
-
"count": count,
383
-
"can_create_bundle": count >= BUNDLE_SIZE,
384
-
"target_bundle": m.targetBundle,
385
-
"min_timestamp": m.minTimestamp,
386
-
"validated": m.validated,
387
-
}
388
-
389
-
if count > 0 {
390
-
stats["first_time"] = m.operations[0].CreatedAt
391
-
stats["last_time"] = m.operations[len(m.operations)-1].CreatedAt
392
-
393
-
// Calculate size and unique DIDs
394
-
totalSize := 0
395
-
didSet := make(map[string]bool)
396
-
for _, op := range m.operations {
397
-
totalSize += len(op.RawJSON)
398
-
didSet[op.DID] = true
399
-
}
400
-
stats["size_bytes"] = totalSize
401
-
stats["did_count"] = len(didSet)
402
-
}
403
-
404
-
return stats
405
-
}
406
-
407
-
// Delete removes the mempool file
408
-
func (m *Mempool) Delete() error {
409
-
if err := os.Remove(m.file); err != nil && !os.IsNotExist(err) {
410
-
return fmt.Errorf("failed to delete mempool file: %w", err)
411
-
}
412
-
return nil
413
-
}
414
-
415
-
// GetFilename returns the mempool filename
416
-
func (m *Mempool) GetFilename() string {
417
-
return filepath.Base(m.file)
418
-
}
···
bundle/metadata.go
internal/bundle/metadata.go
bundle/metadata.go
internal/bundle/metadata.go
+6
-16
bundle/types.go
internal/bundle/types.go
+6
-16
bundle/types.go
internal/bundle/types.go
···
5
"path/filepath"
6
"time"
7
8
"tangled.org/atscan.net/plcbundle/plcclient"
9
-
)
10
-
11
-
const (
12
-
// BUNDLE_SIZE is the standard number of operations per bundle
13
-
BUNDLE_SIZE = 10000
14
)
15
16
// Bundle represents a PLC bundle
···
49
if len(b.Operations) > 0 {
50
return len(b.Operations)
51
}
52
-
return BUNDLE_SIZE
53
}
54
55
// CompressionRatio returns the compression ratio
···
65
if b.BundleNumber < 1 {
66
return fmt.Errorf("invalid bundle number: %d", b.BundleNumber)
67
}
68
-
if len(b.Operations) != BUNDLE_SIZE {
69
-
return fmt.Errorf("invalid operation count: expected %d, got %d", BUNDLE_SIZE, len(b.Operations))
70
}
71
if b.StartTime.After(b.EndTime) {
72
return fmt.Errorf("start_time is after end_time")
···
163
IndexUpdated bool
164
}
165
166
-
// Logger interface for bundle operations
167
-
type Logger interface {
168
-
Printf(format string, v ...interface{})
169
-
Println(v ...interface{})
170
-
}
171
-
172
// Config holds configuration for bundle operations
173
type Config struct {
174
BundleDir string
···
176
AutoRebuild bool
177
RebuildWorkers int // Number of workers for parallel rebuild (0 = auto-detect)
178
RebuildProgress func(current, total int) // Progress callback for rebuild
179
-
Logger Logger
180
}
181
182
// DefaultConfig returns default configuration
···
199
ProgressFunc func(downloaded, total int, bytesDownloaded, bytesTotal int64)
200
SaveInterval time.Duration
201
Verbose bool
202
-
Logger Logger
203
}
204
205
// CloneResult contains cloning results
···
5
"path/filepath"
6
"time"
7
8
+
"tangled.org/atscan.net/plcbundle/internal/types"
9
"tangled.org/atscan.net/plcbundle/plcclient"
10
)
11
12
// Bundle represents a PLC bundle
···
45
if len(b.Operations) > 0 {
46
return len(b.Operations)
47
}
48
+
return types.BUNDLE_SIZE
49
}
50
51
// CompressionRatio returns the compression ratio
···
61
if b.BundleNumber < 1 {
62
return fmt.Errorf("invalid bundle number: %d", b.BundleNumber)
63
}
64
+
if len(b.Operations) != types.BUNDLE_SIZE {
65
+
return fmt.Errorf("invalid operation count: expected %d, got %d", types.BUNDLE_SIZE, len(b.Operations))
66
}
67
if b.StartTime.After(b.EndTime) {
68
return fmt.Errorf("start_time is after end_time")
···
159
IndexUpdated bool
160
}
161
162
// Config holds configuration for bundle operations
163
type Config struct {
164
BundleDir string
···
166
AutoRebuild bool
167
RebuildWorkers int // Number of workers for parallel rebuild (0 = auto-detect)
168
RebuildProgress func(current, total int) // Progress callback for rebuild
169
+
Logger types.Logger
170
}
171
172
// DefaultConfig returns default configuration
···
189
ProgressFunc func(downloaded, total int, bytesDownloaded, bytesTotal int64)
190
SaveInterval time.Duration
191
Verbose bool
192
+
Logger types.Logger
193
}
194
195
// CloneResult contains cloning results
+1
-2
cmd/plcbundle/compare.go
+1
-2
cmd/plcbundle/compare.go
+7
-6
cmd/plcbundle/info.go
+7
-6
cmd/plcbundle/info.go
···
9
"strings"
10
"time"
11
12
-
"tangled.org/atscan.net/plcbundle/bundle"
13
)
14
15
func showGeneralInfo(mgr *bundle.Manager, dir string, verbose bool, showBundles bool, verify bool, showTimeline bool) {
···
83
// Operations count (exact calculation)
84
mempoolStats := mgr.GetMempoolStats()
85
mempoolCount := mempoolStats["count"].(int)
86
-
bundleOpsCount := bundleCount * bundle.BUNDLE_SIZE
87
totalOps := bundleOpsCount + mempoolCount
88
89
fmt.Printf("🔢 Operations\n")
···
146
if mempoolCount > 0 {
147
targetBundle := mempoolStats["target_bundle"].(int)
148
canCreate := mempoolStats["can_create_bundle"].(bool)
149
-
progress := float64(mempoolCount) / float64(bundle.BUNDLE_SIZE) * 100
150
151
fmt.Printf("🔄 Mempool (next bundle: %06d)\n", targetBundle)
152
-
fmt.Printf(" Operations: %s / %s\n", formatNumber(mempoolCount), formatNumber(bundle.BUNDLE_SIZE))
153
fmt.Printf(" Progress: %.1f%%\n", progress)
154
155
// Progress bar
156
barWidth := 40
157
-
filled := int(float64(barWidth) * float64(mempoolCount) / float64(bundle.BUNDLE_SIZE))
158
if filled > barWidth {
159
filled = barWidth
160
}
···
164
if canCreate {
165
fmt.Printf(" ✓ Ready to create bundle\n")
166
} else {
167
-
remaining := bundle.BUNDLE_SIZE - mempoolCount
168
fmt.Printf(" Need %s more operations\n", formatNumber(remaining))
169
}
170
fmt.Printf("\n")
···
9
"strings"
10
"time"
11
12
+
"tangled.org/atscan.net/plcbundle/internal/bundle"
13
+
"tangled.org/atscan.net/plcbundle/internal/types"
14
)
15
16
func showGeneralInfo(mgr *bundle.Manager, dir string, verbose bool, showBundles bool, verify bool, showTimeline bool) {
···
84
// Operations count (exact calculation)
85
mempoolStats := mgr.GetMempoolStats()
86
mempoolCount := mempoolStats["count"].(int)
87
+
bundleOpsCount := bundleCount * types.BUNDLE_SIZE
88
totalOps := bundleOpsCount + mempoolCount
89
90
fmt.Printf("🔢 Operations\n")
···
147
if mempoolCount > 0 {
148
targetBundle := mempoolStats["target_bundle"].(int)
149
canCreate := mempoolStats["can_create_bundle"].(bool)
150
+
progress := float64(mempoolCount) / float64(types.BUNDLE_SIZE) * 100
151
152
fmt.Printf("🔄 Mempool (next bundle: %06d)\n", targetBundle)
153
+
fmt.Printf(" Operations: %s / %s\n", formatNumber(mempoolCount), formatNumber(types.BUNDLE_SIZE))
154
fmt.Printf(" Progress: %.1f%%\n", progress)
155
156
// Progress bar
157
barWidth := 40
158
+
filled := int(float64(barWidth) * float64(mempoolCount) / float64(types.BUNDLE_SIZE))
159
if filled > barWidth {
160
filled = barWidth
161
}
···
165
if canCreate {
166
fmt.Printf(" ✓ Ready to create bundle\n")
167
} else {
168
+
remaining := types.BUNDLE_SIZE - mempoolCount
169
fmt.Printf(" Need %s more operations\n", formatNumber(remaining))
170
}
171
fmt.Printf("\n")
+6
-5
cmd/plcbundle/main.go
+6
-5
cmd/plcbundle/main.go
···
18
19
"github.com/goccy/go-json"
20
21
-
"tangled.org/atscan.net/plcbundle/bundle"
22
"tangled.org/atscan.net/plcbundle/internal/didindex"
23
"tangled.org/atscan.net/plcbundle/plcclient"
24
)
25
···
1165
fmt.Printf("Mempool Status:\n")
1166
fmt.Printf(" Target bundle: %06d\n", targetBundle)
1167
fmt.Printf(" Operations: %d\n", count)
1168
-
fmt.Printf(" Can create bundle: %v (need %d)\n", canCreate, bundle.BUNDLE_SIZE)
1169
fmt.Printf(" Min timestamp: %s\n", minTimestamp.Format("2006-01-02 15:04:05"))
1170
1171
validationIcon := "✓"
···
1187
fmt.Printf(" Last operation: %s\n", lastTime.Format("2006-01-02 15:04:05"))
1188
}
1189
1190
-
progress := float64(count) / float64(bundle.BUNDLE_SIZE) * 100
1191
-
fmt.Printf(" Progress: %.1f%% (%d/%d)\n", progress, count, bundle.BUNDLE_SIZE)
1192
1193
// Show progress bar
1194
barWidth := 40
1195
-
filled := int(float64(barWidth) * float64(count) / float64(bundle.BUNDLE_SIZE))
1196
if filled > barWidth {
1197
filled = barWidth
1198
}
···
18
19
"github.com/goccy/go-json"
20
21
+
"tangled.org/atscan.net/plcbundle/internal/bundle"
22
"tangled.org/atscan.net/plcbundle/internal/didindex"
23
+
"tangled.org/atscan.net/plcbundle/internal/types"
24
"tangled.org/atscan.net/plcbundle/plcclient"
25
)
26
···
1166
fmt.Printf("Mempool Status:\n")
1167
fmt.Printf(" Target bundle: %06d\n", targetBundle)
1168
fmt.Printf(" Operations: %d\n", count)
1169
+
fmt.Printf(" Can create bundle: %v (need %d)\n", canCreate, types.BUNDLE_SIZE)
1170
fmt.Printf(" Min timestamp: %s\n", minTimestamp.Format("2006-01-02 15:04:05"))
1171
1172
validationIcon := "✓"
···
1188
fmt.Printf(" Last operation: %s\n", lastTime.Format("2006-01-02 15:04:05"))
1189
}
1190
1191
+
progress := float64(count) / float64(types.BUNDLE_SIZE) * 100
1192
+
fmt.Printf(" Progress: %.1f%% (%d/%d)\n", progress, count, types.BUNDLE_SIZE)
1193
1194
// Show progress bar
1195
barWidth := 40
1196
+
filled := int(float64(barWidth) * float64(count) / float64(types.BUNDLE_SIZE))
1197
if filled > barWidth {
1198
filled = barWidth
1199
}
+17
-16
cmd/plcbundle/server.go
+17
-16
cmd/plcbundle/server.go
···
15
"github.com/goccy/go-json"
16
"github.com/gorilla/websocket"
17
18
-
"tangled.org/atscan.net/plcbundle/bundle"
19
"tangled.org/atscan.net/plcbundle/plcclient"
20
)
21
···
256
sb.WriteString("\nMempool Stats\n")
257
sb.WriteString("━━━━━━━━━━━━━\n")
258
sb.WriteString(fmt.Sprintf(" Target bundle: %d\n", targetBundle))
259
-
sb.WriteString(fmt.Sprintf(" Operations: %d / %d\n", count, bundle.BUNDLE_SIZE))
260
sb.WriteString(fmt.Sprintf(" Can create bundle: %v\n", canCreate))
261
262
if count > 0 {
263
-
progress := float64(count) / float64(bundle.BUNDLE_SIZE) * 100
264
sb.WriteString(fmt.Sprintf(" Progress: %.1f%%\n", progress))
265
266
barWidth := 50
267
-
filled := int(float64(barWidth) * float64(count) / float64(bundle.BUNDLE_SIZE))
268
if filled > barWidth {
269
filled = barWidth
270
}
···
355
sb.WriteString(" Default: starts from latest (skips all historical data)\n")
356
357
latestCursor := mgr.GetCurrentCursor()
358
-
bundledOps := len(index.GetBundles()) * bundle.BUNDLE_SIZE
359
mempoolOps := latestCursor - bundledOps
360
361
if syncMode && mempoolOps > 0 {
···
517
}
518
}
519
520
-
totalOps := bundleCount * bundle.BUNDLE_SIZE
521
response.Bundles.TotalOperations = totalOps
522
523
duration := response.Bundles.EndTime.Sub(response.Bundles.StartTime)
···
536
CanCreateBundle: mempoolStats["can_create_bundle"].(bool),
537
MinTimestamp: mempoolStats["min_timestamp"].(time.Time),
538
Validated: mempoolStats["validated"].(bool),
539
-
ProgressPercent: float64(count) / float64(bundle.BUNDLE_SIZE) * 100,
540
-
BundleSize: bundle.BUNDLE_SIZE,
541
-
OperationsNeeded: bundle.BUNDLE_SIZE - count,
542
}
543
544
if firstTime, ok := mempoolStats["first_time"].(time.Time); ok {
···
550
mempool.LastOpAgeSeconds = int(time.Since(lastTime).Seconds())
551
}
552
553
-
if count > 100 && count < bundle.BUNDLE_SIZE {
554
if !mempool.FirstTime.IsZero() && !mempool.LastTime.IsZero() {
555
timespan := mempool.LastTime.Sub(mempool.FirstTime)
556
if timespan.Seconds() > 0 {
557
opsPerSec := float64(count) / timespan.Seconds()
558
-
remaining := bundle.BUNDLE_SIZE - count
559
mempool.EtaNextBundleSeconds = int(float64(remaining) / opsPerSec)
560
}
561
}
···
770
currentRecord := startCursor
771
772
if len(bundles) > 0 {
773
-
startBundleIdx := startCursor / bundle.BUNDLE_SIZE
774
-
startPosition := startCursor % bundle.BUNDLE_SIZE
775
776
if startBundleIdx < len(bundles) {
777
for i := startBundleIdx; i < len(bundles); i++ {
···
790
}
791
792
lastSeenMempoolCount := 0
793
-
if err := streamMempool(conn, mgr, startCursor, len(bundles)*bundle.BUNDLE_SIZE, ¤tRecord, &lastSeenMempoolCount, done); err != nil {
794
return err
795
}
796
···
821
fmt.Fprintf(os.Stderr, "WebSocket: %d new bundle(s) created (operations already streamed from mempool)\n", newBundleCount)
822
}
823
824
-
currentRecord += newBundleCount * bundle.BUNDLE_SIZE
825
lastBundleCount = len(bundles)
826
lastSeenMempoolCount = 0
827
}
828
829
-
if err := streamMempool(conn, mgr, startCursor, len(bundles)*bundle.BUNDLE_SIZE, ¤tRecord, &lastSeenMempoolCount, done); err != nil {
830
return err
831
}
832
···
15
"github.com/goccy/go-json"
16
"github.com/gorilla/websocket"
17
18
+
"tangled.org/atscan.net/plcbundle/internal/bundle"
19
+
"tangled.org/atscan.net/plcbundle/internal/types"
20
"tangled.org/atscan.net/plcbundle/plcclient"
21
)
22
···
257
sb.WriteString("\nMempool Stats\n")
258
sb.WriteString("━━━━━━━━━━━━━\n")
259
sb.WriteString(fmt.Sprintf(" Target bundle: %d\n", targetBundle))
260
+
sb.WriteString(fmt.Sprintf(" Operations: %d / %d\n", count, types.BUNDLE_SIZE))
261
sb.WriteString(fmt.Sprintf(" Can create bundle: %v\n", canCreate))
262
263
if count > 0 {
264
+
progress := float64(count) / float64(types.BUNDLE_SIZE) * 100
265
sb.WriteString(fmt.Sprintf(" Progress: %.1f%%\n", progress))
266
267
barWidth := 50
268
+
filled := int(float64(barWidth) * float64(count) / float64(types.BUNDLE_SIZE))
269
if filled > barWidth {
270
filled = barWidth
271
}
···
356
sb.WriteString(" Default: starts from latest (skips all historical data)\n")
357
358
latestCursor := mgr.GetCurrentCursor()
359
+
bundledOps := len(index.GetBundles()) * types.BUNDLE_SIZE
360
mempoolOps := latestCursor - bundledOps
361
362
if syncMode && mempoolOps > 0 {
···
518
}
519
}
520
521
+
totalOps := bundleCount * types.BUNDLE_SIZE
522
response.Bundles.TotalOperations = totalOps
523
524
duration := response.Bundles.EndTime.Sub(response.Bundles.StartTime)
···
537
CanCreateBundle: mempoolStats["can_create_bundle"].(bool),
538
MinTimestamp: mempoolStats["min_timestamp"].(time.Time),
539
Validated: mempoolStats["validated"].(bool),
540
+
ProgressPercent: float64(count) / float64(types.BUNDLE_SIZE) * 100,
541
+
BundleSize: types.BUNDLE_SIZE,
542
+
OperationsNeeded: types.BUNDLE_SIZE - count,
543
}
544
545
if firstTime, ok := mempoolStats["first_time"].(time.Time); ok {
···
551
mempool.LastOpAgeSeconds = int(time.Since(lastTime).Seconds())
552
}
553
554
+
if count > 100 && count < types.BUNDLE_SIZE {
555
if !mempool.FirstTime.IsZero() && !mempool.LastTime.IsZero() {
556
timespan := mempool.LastTime.Sub(mempool.FirstTime)
557
if timespan.Seconds() > 0 {
558
opsPerSec := float64(count) / timespan.Seconds()
559
+
remaining := types.BUNDLE_SIZE - count
560
mempool.EtaNextBundleSeconds = int(float64(remaining) / opsPerSec)
561
}
562
}
···
771
currentRecord := startCursor
772
773
if len(bundles) > 0 {
774
+
startBundleIdx := startCursor / types.BUNDLE_SIZE
775
+
startPosition := startCursor % types.BUNDLE_SIZE
776
777
if startBundleIdx < len(bundles) {
778
for i := startBundleIdx; i < len(bundles); i++ {
···
791
}
792
793
lastSeenMempoolCount := 0
794
+
if err := streamMempool(conn, mgr, startCursor, len(bundles)*types.BUNDLE_SIZE, ¤tRecord, &lastSeenMempoolCount, done); err != nil {
795
return err
796
}
797
···
822
fmt.Fprintf(os.Stderr, "WebSocket: %d new bundle(s) created (operations already streamed from mempool)\n", newBundleCount)
823
}
824
825
+
currentRecord += newBundleCount * types.BUNDLE_SIZE
826
lastBundleCount = len(bundles)
827
lastSeenMempoolCount = 0
828
}
829
830
+
if err := streamMempool(conn, mgr, startCursor, len(bundles)*types.BUNDLE_SIZE, ¤tRecord, &lastSeenMempoolCount, done); err != nil {
831
return err
832
}
833
+1
-1
detector/runner.go
+1
-1
detector/runner.go
+418
internal/mempool/mempool.go
+418
internal/mempool/mempool.go
···
1
package mempool
2
+
3
+
import (
4
+
"bufio"
5
+
"bytes"
6
+
"fmt"
7
+
"os"
8
+
"path/filepath"
9
+
"sync"
10
+
"time"
11
+
12
+
"github.com/goccy/go-json"
13
+
"tangled.org/atscan.net/plcbundle/internal/types"
14
+
"tangled.org/atscan.net/plcbundle/plcclient"
15
+
)
16
+
17
+
const MEMPOOL_FILE_PREFIX = "plc_mempool_"
18
+
19
+
// Mempool stores operations waiting to be bundled
20
+
// Operations must be strictly chronological
21
+
type Mempool struct {
22
+
operations []plcclient.PLCOperation
23
+
targetBundle int // Which bundle number these operations are for
24
+
minTimestamp time.Time // Operations must be after this time
25
+
file string
26
+
mu sync.RWMutex
27
+
logger types.Logger
28
+
validated bool // Track if we've validated chronological order
29
+
dirty bool // Track if mempool changed
30
+
}
31
+
32
+
// NewMempool creates a new mempool for a specific bundle number
33
+
func NewMempool(bundleDir string, targetBundle int, minTimestamp time.Time, logger types.Logger) (*Mempool, error) {
34
+
filename := fmt.Sprintf("%s%06d.jsonl", MEMPOOL_FILE_PREFIX, targetBundle)
35
+
36
+
m := &Mempool{
37
+
file: filepath.Join(bundleDir, filename),
38
+
targetBundle: targetBundle,
39
+
minTimestamp: minTimestamp,
40
+
operations: make([]plcclient.PLCOperation, 0),
41
+
logger: logger,
42
+
validated: false,
43
+
}
44
+
45
+
// Load existing mempool from disk if it exists
46
+
if err := m.Load(); err != nil {
47
+
// If file doesn't exist, that's OK
48
+
if !os.IsNotExist(err) {
49
+
return nil, fmt.Errorf("failed to load mempool: %w", err)
50
+
}
51
+
}
52
+
53
+
return m, nil
54
+
}
55
+
56
+
// Add adds operations to the mempool with strict validation
57
+
func (m *Mempool) Add(ops []plcclient.PLCOperation) (int, error) {
58
+
m.mu.Lock()
59
+
defer m.mu.Unlock()
60
+
61
+
if len(ops) == 0 {
62
+
return 0, nil
63
+
}
64
+
65
+
// Build existing CID set
66
+
existingCIDs := make(map[string]bool)
67
+
for _, op := range m.operations {
68
+
existingCIDs[op.CID] = true
69
+
}
70
+
71
+
// Validate and add operations
72
+
var newOps []plcclient.PLCOperation
73
+
var lastTime time.Time
74
+
75
+
// Start from last operation time if we have any
76
+
if len(m.operations) > 0 {
77
+
lastTime = m.operations[len(m.operations)-1].CreatedAt
78
+
} else {
79
+
lastTime = m.minTimestamp
80
+
}
81
+
82
+
for _, op := range ops {
83
+
// Skip duplicates
84
+
if existingCIDs[op.CID] {
85
+
continue
86
+
}
87
+
88
+
// CRITICAL: Validate chronological order
89
+
if !op.CreatedAt.After(lastTime) && !op.CreatedAt.Equal(lastTime) {
90
+
return len(newOps), fmt.Errorf(
91
+
"chronological violation: operation %s at %s is not after %s",
92
+
op.CID, op.CreatedAt.Format(time.RFC3339Nano), lastTime.Format(time.RFC3339Nano),
93
+
)
94
+
}
95
+
96
+
// Validate operation is after minimum timestamp
97
+
if op.CreatedAt.Before(m.minTimestamp) {
98
+
return len(newOps), fmt.Errorf(
99
+
"operation %s at %s is before minimum timestamp %s (belongs in earlier bundle)",
100
+
op.CID, op.CreatedAt.Format(time.RFC3339Nano), m.minTimestamp.Format(time.RFC3339Nano),
101
+
)
102
+
}
103
+
104
+
newOps = append(newOps, op)
105
+
existingCIDs[op.CID] = true
106
+
lastTime = op.CreatedAt
107
+
}
108
+
109
+
// Add new operations
110
+
m.operations = append(m.operations, newOps...)
111
+
m.validated = true
112
+
m.dirty = true
113
+
114
+
return len(newOps), nil
115
+
}
116
+
117
+
// Validate performs a full chronological validation of all operations
118
+
func (m *Mempool) Validate() error {
119
+
m.mu.RLock()
120
+
defer m.mu.RUnlock()
121
+
122
+
if len(m.operations) == 0 {
123
+
return nil
124
+
}
125
+
126
+
// Check all operations are after minimum timestamp
127
+
for i, op := range m.operations {
128
+
if op.CreatedAt.Before(m.minTimestamp) {
129
+
return fmt.Errorf(
130
+
"operation %d (CID: %s) at %s is before minimum timestamp %s",
131
+
i, op.CID, op.CreatedAt.Format(time.RFC3339Nano), m.minTimestamp.Format(time.RFC3339Nano),
132
+
)
133
+
}
134
+
}
135
+
136
+
// Check chronological order
137
+
for i := 1; i < len(m.operations); i++ {
138
+
prev := m.operations[i-1]
139
+
curr := m.operations[i]
140
+
141
+
if curr.CreatedAt.Before(prev.CreatedAt) {
142
+
return fmt.Errorf(
143
+
"chronological violation at index %d: %s (%s) is before %s (%s)",
144
+
i, curr.CID, curr.CreatedAt.Format(time.RFC3339Nano),
145
+
prev.CID, prev.CreatedAt.Format(time.RFC3339Nano),
146
+
)
147
+
}
148
+
}
149
+
150
+
// Check for duplicate CIDs
151
+
cidSet := make(map[string]int)
152
+
for i, op := range m.operations {
153
+
if prevIdx, exists := cidSet[op.CID]; exists {
154
+
return fmt.Errorf(
155
+
"duplicate CID %s at indices %d and %d",
156
+
op.CID, prevIdx, i,
157
+
)
158
+
}
159
+
cidSet[op.CID] = i
160
+
}
161
+
162
+
return nil
163
+
}
164
+
165
+
// Count returns the number of operations in mempool
166
+
func (m *Mempool) Count() int {
167
+
m.mu.RLock()
168
+
defer m.mu.RUnlock()
169
+
return len(m.operations)
170
+
}
171
+
172
+
// Take removes and returns up to n operations from the front
173
+
func (m *Mempool) Take(n int) ([]plcclient.PLCOperation, error) {
174
+
m.mu.Lock()
175
+
defer m.mu.Unlock()
176
+
177
+
// Validate before taking
178
+
if err := m.validateLocked(); err != nil {
179
+
return nil, fmt.Errorf("mempool validation failed: %w", err)
180
+
}
181
+
182
+
if n > len(m.operations) {
183
+
n = len(m.operations)
184
+
}
185
+
186
+
result := make([]plcclient.PLCOperation, n)
187
+
copy(result, m.operations[:n])
188
+
189
+
// Remove taken operations
190
+
m.operations = m.operations[n:]
191
+
192
+
return result, nil
193
+
}
194
+
195
+
// validateLocked performs validation with lock already held
196
+
func (m *Mempool) validateLocked() error {
197
+
if m.validated {
198
+
return nil
199
+
}
200
+
201
+
if len(m.operations) == 0 {
202
+
return nil
203
+
}
204
+
205
+
// Check chronological order
206
+
lastTime := m.minTimestamp
207
+
for i, op := range m.operations {
208
+
if op.CreatedAt.Before(lastTime) {
209
+
return fmt.Errorf(
210
+
"chronological violation at index %d: %s is before %s",
211
+
i, op.CreatedAt.Format(time.RFC3339Nano), lastTime.Format(time.RFC3339Nano),
212
+
)
213
+
}
214
+
lastTime = op.CreatedAt
215
+
}
216
+
217
+
m.validated = true
218
+
return nil
219
+
}
220
+
221
+
// Peek returns up to n operations without removing them
222
+
func (m *Mempool) Peek(n int) []plcclient.PLCOperation {
223
+
m.mu.RLock()
224
+
defer m.mu.RUnlock()
225
+
226
+
if n > len(m.operations) {
227
+
n = len(m.operations)
228
+
}
229
+
230
+
result := make([]plcclient.PLCOperation, n)
231
+
copy(result, m.operations[:n])
232
+
233
+
return result
234
+
}
235
+
236
+
// Clear removes all operations
237
+
func (m *Mempool) Clear() {
238
+
m.mu.Lock()
239
+
defer m.mu.Unlock()
240
+
m.operations = make([]plcclient.PLCOperation, 0)
241
+
m.validated = false
242
+
}
243
+
244
+
// Save persists mempool to disk
245
+
func (m *Mempool) Save() error {
246
+
m.mu.Lock()
247
+
defer m.mu.Unlock()
248
+
249
+
if !m.dirty {
250
+
return nil
251
+
}
252
+
253
+
if len(m.operations) == 0 {
254
+
// Remove file if empty
255
+
os.Remove(m.file)
256
+
return nil
257
+
}
258
+
259
+
// Validate before saving
260
+
if err := m.validateLocked(); err != nil {
261
+
return fmt.Errorf("mempool validation failed, refusing to save: %w", err)
262
+
}
263
+
264
+
// Serialize to JSONL
265
+
var buf bytes.Buffer
266
+
for _, op := range m.operations {
267
+
if len(op.RawJSON) > 0 {
268
+
buf.Write(op.RawJSON)
269
+
} else {
270
+
data, _ := json.Marshal(op)
271
+
buf.Write(data)
272
+
}
273
+
buf.WriteByte('\n')
274
+
}
275
+
276
+
// Write atomically
277
+
tempFile := m.file + ".tmp"
278
+
if err := os.WriteFile(tempFile, buf.Bytes(), 0644); err != nil {
279
+
return fmt.Errorf("failed to write mempool: %w", err)
280
+
}
281
+
282
+
if err := os.Rename(tempFile, m.file); err != nil {
283
+
os.Remove(tempFile)
284
+
return fmt.Errorf("failed to rename mempool file: %w", err)
285
+
}
286
+
287
+
m.dirty = false
288
+
return nil
289
+
}
290
+
291
+
// Load reads mempool from disk and validates it
292
+
func (m *Mempool) Load() error {
293
+
data, err := os.ReadFile(m.file)
294
+
if err != nil {
295
+
return err
296
+
}
297
+
298
+
m.mu.Lock()
299
+
defer m.mu.Unlock()
300
+
301
+
// Parse JSONL
302
+
scanner := bufio.NewScanner(bytes.NewReader(data))
303
+
buf := make([]byte, 0, 64*1024)
304
+
scanner.Buffer(buf, 1024*1024)
305
+
306
+
m.operations = make([]plcclient.PLCOperation, 0)
307
+
308
+
for scanner.Scan() {
309
+
line := scanner.Bytes()
310
+
if len(line) == 0 {
311
+
continue
312
+
}
313
+
314
+
var op plcclient.PLCOperation
315
+
if err := json.Unmarshal(line, &op); err != nil {
316
+
return fmt.Errorf("failed to parse mempool operation: %w", err)
317
+
}
318
+
319
+
op.RawJSON = make([]byte, len(line))
320
+
copy(op.RawJSON, line)
321
+
322
+
m.operations = append(m.operations, op)
323
+
}
324
+
325
+
if err := scanner.Err(); err != nil {
326
+
return fmt.Errorf("scanner error: %w", err)
327
+
}
328
+
329
+
// CRITICAL: Validate loaded data
330
+
if err := m.validateLocked(); err != nil {
331
+
return fmt.Errorf("loaded mempool failed validation: %w", err)
332
+
}
333
+
334
+
if len(m.operations) > 0 {
335
+
m.logger.Printf("Loaded %d operations from mempool for bundle %06d", len(m.operations), m.targetBundle)
336
+
}
337
+
338
+
return nil
339
+
}
340
+
341
+
// GetFirstTime returns the created_at of the first operation
342
+
func (m *Mempool) GetFirstTime() string {
343
+
m.mu.RLock()
344
+
defer m.mu.RUnlock()
345
+
346
+
if len(m.operations) == 0 {
347
+
return ""
348
+
}
349
+
350
+
return m.operations[0].CreatedAt.Format(time.RFC3339Nano)
351
+
}
352
+
353
+
// GetLastTime returns the created_at of the last operation
354
+
func (m *Mempool) GetLastTime() string {
355
+
m.mu.RLock()
356
+
defer m.mu.RUnlock()
357
+
358
+
if len(m.operations) == 0 {
359
+
return ""
360
+
}
361
+
362
+
return m.operations[len(m.operations)-1].CreatedAt.Format(time.RFC3339Nano)
363
+
}
364
+
365
+
// GetTargetBundle returns the bundle number this mempool is for
366
+
func (m *Mempool) GetTargetBundle() int {
367
+
return m.targetBundle
368
+
}
369
+
370
+
// GetMinTimestamp returns the minimum timestamp for operations
371
+
func (m *Mempool) GetMinTimestamp() time.Time {
372
+
return m.minTimestamp
373
+
}
374
+
375
+
// Stats returns mempool statistics
376
+
func (m *Mempool) Stats() map[string]interface{} {
377
+
m.mu.RLock()
378
+
defer m.mu.RUnlock()
379
+
380
+
count := len(m.operations)
381
+
382
+
stats := map[string]interface{}{
383
+
"count": count,
384
+
"can_create_bundle": count >= types.BUNDLE_SIZE,
385
+
"target_bundle": m.targetBundle,
386
+
"min_timestamp": m.minTimestamp,
387
+
"validated": m.validated,
388
+
}
389
+
390
+
if count > 0 {
391
+
stats["first_time"] = m.operations[0].CreatedAt
392
+
stats["last_time"] = m.operations[len(m.operations)-1].CreatedAt
393
+
394
+
// Calculate size and unique DIDs
395
+
totalSize := 0
396
+
didSet := make(map[string]bool)
397
+
for _, op := range m.operations {
398
+
totalSize += len(op.RawJSON)
399
+
didSet[op.DID] = true
400
+
}
401
+
stats["size_bytes"] = totalSize
402
+
stats["did_count"] = len(didSet)
403
+
}
404
+
405
+
return stats
406
+
}
407
+
408
+
// Delete removes the mempool file
409
+
func (m *Mempool) Delete() error {
410
+
if err := os.Remove(m.file); err != nil && !os.IsNotExist(err) {
411
+
return fmt.Errorf("failed to delete mempool file: %w", err)
412
+
}
413
+
return nil
414
+
}
415
+
416
+
// GetFilename returns the mempool filename
417
+
func (m *Mempool) GetFilename() string {
418
+
return filepath.Base(m.file)
419
+
}
+18
internal/types/types.go
+18
internal/types/types.go
···
···
1
+
package types
2
+
3
+
// Logger is a simple logging interface used throughout plcbundle
4
+
type Logger interface {
5
+
Printf(format string, v ...interface{})
6
+
Println(v ...interface{})
7
+
}
8
+
9
+
const (
10
+
// BUNDLE_SIZE is the standard number of operations per bundle
11
+
BUNDLE_SIZE = 10000
12
+
13
+
// INDEX_FILE is the default index filename
14
+
INDEX_FILE = "plc_bundles.json"
15
+
16
+
// INDEX_VERSION is the current index format version
17
+
INDEX_VERSION = "1.0"
18
+
)
-148
plcbundle.go
-148
plcbundle.go
···
1
-
package plcbundle
2
-
3
-
import (
4
-
"context"
5
-
"io"
6
-
"time"
7
-
8
-
"tangled.org/atscan.net/plcbundle/bundle"
9
-
"tangled.org/atscan.net/plcbundle/plcclient"
10
-
)
11
-
12
-
// Re-export commonly used types for convenience
13
-
type (
14
-
Bundle = bundle.Bundle
15
-
BundleMetadata = bundle.BundleMetadata
16
-
Index = bundle.Index
17
-
Manager = bundle.Manager
18
-
Config = bundle.Config
19
-
VerificationResult = bundle.VerificationResult
20
-
ChainVerificationResult = bundle.ChainVerificationResult
21
-
DirectoryScanResult = bundle.DirectoryScanResult
22
-
Logger = bundle.Logger
23
-
24
-
PLCOperation = plcclient.PLCOperation
25
-
PLCClient = plcclient.Client
26
-
ExportOptions = plcclient.ExportOptions
27
-
)
28
-
29
-
// Re-export constants
30
-
const (
31
-
BUNDLE_SIZE = bundle.BUNDLE_SIZE
32
-
INDEX_FILE = bundle.INDEX_FILE
33
-
)
34
-
35
-
// NewManager creates a new bundle manager (convenience wrapper)
36
-
func NewManager(config *Config, plcClient *PLCClient) (*Manager, error) {
37
-
return bundle.NewManager(config, plcClient)
38
-
}
39
-
40
-
// NewPLCClient creates a new PLC client (convenience wrapper)
41
-
func NewPLCClient(baseURL string, opts ...plcclient.ClientOption) *PLCClient {
42
-
return plcclient.NewClient(baseURL, opts...)
43
-
}
44
-
45
-
// DefaultConfig returns default configuration (convenience wrapper)
46
-
func DefaultConfig(bundleDir string) *Config {
47
-
return bundle.DefaultConfig(bundleDir)
48
-
}
49
-
50
-
// NewIndex creates a new empty index (convenience wrapper)
51
-
func NewIndex(origin string) *Index {
52
-
return bundle.NewIndex(origin)
53
-
}
54
-
55
-
// LoadIndex loads an index from a file (convenience wrapper)
56
-
func LoadIndex(path string) (*Index, error) {
57
-
return bundle.LoadIndex(path)
58
-
}
59
-
60
-
// BundleManager provides a high-level API for bundle operations
61
-
type BundleManager struct {
62
-
mgr *Manager
63
-
}
64
-
65
-
// New creates a new BundleManager with default settings
66
-
func New(bundleDir string, plcURL string) (*BundleManager, error) {
67
-
config := DefaultConfig(bundleDir)
68
-
var plcClient *PLCClient
69
-
if plcURL != "" {
70
-
plcClient = NewPLCClient(plcURL)
71
-
}
72
-
73
-
mgr, err := NewManager(config, plcClient)
74
-
if err != nil {
75
-
return nil, err
76
-
}
77
-
78
-
return &BundleManager{mgr: mgr}, nil
79
-
}
80
-
81
-
// Close closes the manager
82
-
func (bm *BundleManager) Close() {
83
-
bm.mgr.Close()
84
-
}
85
-
86
-
// FetchNext fetches the next bundle from PLC
87
-
func (bm *BundleManager) FetchNext(ctx context.Context) (*Bundle, error) {
88
-
b, err := bm.mgr.FetchNextBundle(ctx, false)
89
-
if err != nil {
90
-
return nil, err
91
-
}
92
-
return b, bm.mgr.SaveBundle(ctx, b, false)
93
-
}
94
-
95
-
// Load loads a bundle by number
96
-
func (bm *BundleManager) Load(ctx context.Context, bundleNumber int) (*Bundle, error) {
97
-
return bm.mgr.LoadBundle(ctx, bundleNumber)
98
-
}
99
-
100
-
// Verify verifies a bundle
101
-
func (bm *BundleManager) Verify(ctx context.Context, bundleNumber int) (*VerificationResult, error) {
102
-
return bm.mgr.VerifyBundle(ctx, bundleNumber)
103
-
}
104
-
105
-
// VerifyChain verifies the entire chain
106
-
func (bm *BundleManager) VerifyChain(ctx context.Context) (*ChainVerificationResult, error) {
107
-
return bm.mgr.VerifyChain(ctx)
108
-
}
109
-
110
-
// GetIndex returns the index
111
-
func (bm *BundleManager) GetIndex() *Index {
112
-
return bm.mgr.GetIndex()
113
-
}
114
-
115
-
// GetInfo returns manager info
116
-
func (bm *BundleManager) GetInfo() map[string]interface{} {
117
-
return bm.mgr.GetInfo()
118
-
}
119
-
120
-
// Export exports operations from bundles
121
-
func (bm *BundleManager) Export(ctx context.Context, afterTime time.Time, count int) ([]PLCOperation, error) {
122
-
return bm.mgr.ExportOperations(ctx, afterTime, count)
123
-
}
124
-
125
-
// Scan scans the directory and rebuilds the index
126
-
func (bm *BundleManager) Scan() (*DirectoryScanResult, error) {
127
-
return bm.mgr.ScanDirectory()
128
-
}
129
-
130
-
// ScanBundle scans a single bundle file
131
-
func (bm *BundleManager) ScanBundle(path string, bundleNumber int) (*BundleMetadata, error) {
132
-
return bm.mgr.ScanBundle(path, bundleNumber)
133
-
}
134
-
135
-
// IsBundleIndexed checks if a bundle is in the index
136
-
func (bm *BundleManager) IsBundleIndexed(bundleNumber int) bool {
137
-
return bm.mgr.IsBundleIndexed(bundleNumber)
138
-
}
139
-
140
-
// StreamRaw streams raw compressed bundle data
141
-
func (bm *BundleManager) StreamRaw(ctx context.Context, bundleNumber int) (io.ReadCloser, error) {
142
-
return bm.mgr.StreamBundleRaw(ctx, bundleNumber)
143
-
}
144
-
145
-
// StreamDecompressed streams decompressed bundle data
146
-
func (bm *BundleManager) StreamDecompressed(ctx context.Context, bundleNumber int) (io.ReadCloser, error) {
147
-
return bm.mgr.StreamBundleDecompressed(ctx, bundleNumber)
148
-
}
···