atscan.net
[DEPRECATED] Go implementation of plcbundle
try another approach
tree.fail
+18
-4
bundle/manager.go
+18
-4
bundle/manager.go
···
1134
afterTime = lastBundle.EndTime.Format(time.RFC3339Nano)
1135
prevBundleHash = lastBundle.Hash
1136
1137
prevBundle, err := m.LoadBundle(ctx, lastBundle.BundleNumber)
1138
if err == nil {
1139
_, prevBoundaryCIDs = m.operations.GetBoundaryCIDs(prevBundle.Operations)
1140
if !quiet {
1141
-
m.logger.Printf("Previous bundle %06d has %d boundary CIDs at %s",
1142
-
lastBundle.BundleNumber, len(prevBoundaryCIDs), lastBundle.EndTime.Format(time.RFC3339))
1143
}
1144
}
1145
}
1146
1147
-
// Use mempool's last time if available
1148
if m.mempool.Count() > 0 {
1149
mempoolLastTime := m.mempool.GetLastTime()
1150
if mempoolLastTime != "" {
1151
if !quiet {
1152
-
m.logger.Printf("Mempool has %d ops, last at %s", m.mempool.Count(), mempoolLastTime)
1153
}
1154
afterTime = mempoolLastTime
1155
}
1156
}
1157
···
1134
afterTime = lastBundle.EndTime.Format(time.RFC3339Nano)
1135
prevBundleHash = lastBundle.Hash
1136
1137
+
// ALWAYS get boundaries from last bundle initially
1138
prevBundle, err := m.LoadBundle(ctx, lastBundle.BundleNumber)
1139
if err == nil {
1140
_, prevBoundaryCIDs = m.operations.GetBoundaryCIDs(prevBundle.Operations)
1141
if !quiet {
1142
+
m.logger.Printf("Loaded %d boundary CIDs from bundle %06d (at %s)",
1143
+
len(prevBoundaryCIDs), lastBundle.BundleNumber,
1144
+
lastBundle.EndTime.Format(time.RFC3339)[:19])
1145
}
1146
}
1147
}
1148
1149
+
// If mempool has operations, update cursor but KEEP boundaries from bundle
1150
+
// (mempool operations already had boundary dedup applied when they were added)
1151
if m.mempool.Count() > 0 {
1152
mempoolLastTime := m.mempool.GetLastTime()
1153
if mempoolLastTime != "" {
1154
if !quiet {
1155
+
m.logger.Printf("Mempool has %d ops, resuming from %s",
1156
+
m.mempool.Count(), mempoolLastTime[:19])
1157
}
1158
afterTime = mempoolLastTime
1159
+
1160
+
// Calculate boundaries from MEMPOOL for next fetch
1161
+
mempoolOps, _ := m.GetMempoolOperations()
1162
+
if len(mempoolOps) > 0 {
1163
+
_, mempoolBoundaries := m.operations.GetBoundaryCIDs(mempoolOps)
1164
+
prevBoundaryCIDs = mempoolBoundaries
1165
+
if !quiet {
1166
+
m.logger.Printf("Using %d boundary CIDs from mempool", len(prevBoundaryCIDs))
1167
+
}
1168
+
}
1169
}
1170
}
1171
+56
-41
internal/sync/fetcher.go
+56
-41
internal/sync/fetcher.go
···
48
49
seenCIDs := make(map[string]bool)
50
51
-
// Mark previous boundary CIDs as seen
52
-
for cid := range prevBoundaryCIDs {
53
seenCIDs[cid] = true
54
}
55
56
-
if !quiet && len(prevBoundaryCIDs) > 0 {
57
-
f.logger.Printf(" Tracking %d boundary CIDs from previous bundle", len(prevBoundaryCIDs))
58
}
59
60
currentAfter := afterTime
···
112
originalBatchSize := len(batch)
113
totalReceived += originalBatchSize
114
115
-
// ✨ FIX: Collect new ops first (don't mark as seen yet)
116
beforeDedup := len(allNewOps)
117
var batchNewOps []plcclient.PLCOperation
118
119
for _, op := range batch {
120
if !seenCIDs[op.CID] {
121
-
// Collect but don't mark as seen yet
122
batchNewOps = append(batchNewOps, op)
123
}
124
}
125
126
uniqueInBatch := len(batchNewOps)
127
-
dupesFiltered := originalBatchSize - uniqueInBatch
128
-
totalDupes += dupesFiltered
129
130
-
// ✨ FIX: Try to add to mempool BEFORE marking as seen
131
if uniqueInBatch > 0 && mempool != nil {
132
-
added, addErr := mempool.Add(batchNewOps)
133
134
if addErr != nil {
135
-
// ✨ CRITICAL: Add failed - don't mark as seen!
136
if !quiet {
137
f.logger.Printf(" ❌ Mempool add failed: %v", addErr)
138
-
f.logger.Printf(" → %d operations NOT marked as seen (will retry on next sync)", uniqueInBatch)
139
}
140
-
141
-
// Save what we successfully added so far
142
mempool.Save()
143
-
return allNewOps, fetchesMade, fmt.Errorf("mempool add failed for %d operations: %w", uniqueInBatch, addErr)
144
}
145
146
-
// ✨ SUCCESS: Add succeeded, NOW mark as seen and add to result
147
-
if added != uniqueInBatch {
148
-
// Mempool deduplicated some - this is OK, just log
149
-
if !quiet {
150
-
f.logger.Printf(" ℹ️ Mempool deduplicated %d operations (already present)", uniqueInBatch-added)
151
-
}
152
-
}
153
-
154
-
// Mark successfully added ops as seen
155
for _, op := range batchNewOps {
156
seenCIDs[op.CID] = true
157
}
···
159
160
uniqueAdded := len(allNewOps) - beforeDedup
161
162
-
// Show fetch result with running totals
163
if !quiet {
164
opsPerSec := float64(originalBatchSize) / fetchDuration.Seconds()
165
-
166
-
if dupesFiltered > 0 {
167
-
f.logger.Printf(" → +%d unique (%d dupes) in %s • Running: %d/%d (%.0f ops/sec)",
168
-
uniqueAdded, dupesFiltered, fetchDuration, len(allNewOps), target, opsPerSec)
169
} else {
170
f.logger.Printf(" → +%d unique in %s • Running: %d/%d (%.0f ops/sec)",
171
uniqueAdded, fetchDuration, len(allNewOps), target, opsPerSec)
172
}
173
}
174
175
-
// ✨ Save only if threshold met
176
if err := mempool.SaveIfNeeded(); err != nil {
177
f.logger.Printf(" Warning: failed to save mempool: %v", err)
178
}
179
180
-
if !quiet && added > 0 {
181
-
cursor := mempool.GetLastTime()
182
-
f.logger.Printf(" Added to mempool: %d ops (total: %d, cursor: %s)",
183
-
added, mempool.Count(), cursor[:19])
184
-
}
185
} else if uniqueInBatch > 0 {
186
-
// No mempool provided - just collect
187
for _, op := range batchNewOps {
188
seenCIDs[op.CID] = true
189
}
190
allNewOps = append(allNewOps, batchNewOps...)
191
-
}
192
193
-
// Update cursor for next fetch
194
-
if len(batch) > 0 {
195
-
currentAfter = batch[len(batch)-1].CreatedAt.Format(time.RFC3339Nano)
196
}
197
198
-
// Check completeness
199
if originalBatchSize < batchSize {
200
if !quiet {
201
f.logger.Printf(" Incomplete batch (%d/%d) → caught up", originalBatchSize, batchSize)
···
48
49
seenCIDs := make(map[string]bool)
50
51
+
// ✅ Initialize current boundaries from previous bundle (or empty if first fetch)
52
+
currentBoundaryCIDs := prevBoundaryCIDs
53
+
if currentBoundaryCIDs == nil {
54
+
currentBoundaryCIDs = make(map[string]bool)
55
+
}
56
+
57
+
// Mark boundary CIDs as seen to prevent re-inclusion
58
+
for cid := range currentBoundaryCIDs {
59
seenCIDs[cid] = true
60
}
61
62
+
if !quiet && len(currentBoundaryCIDs) > 0 {
63
+
f.logger.Printf(" Starting with %d boundary CIDs from previous bundle", len(currentBoundaryCIDs))
64
}
65
66
currentAfter := afterTime
···
118
originalBatchSize := len(batch)
119
totalReceived += originalBatchSize
120
121
+
// ✅ CRITICAL: Strip boundary duplicates using current boundaries
122
+
batch = f.operations.StripBoundaryDuplicates(
123
+
batch,
124
+
currentAfter,
125
+
currentBoundaryCIDs,
126
+
)
127
+
128
+
afterStripSize := len(batch)
129
+
strippedCount := originalBatchSize - afterStripSize
130
+
131
+
if !quiet && strippedCount > 0 {
132
+
f.logger.Printf(" Stripped %d boundary duplicates from fetch", strippedCount)
133
+
}
134
+
135
+
// Collect new ops (not in seenCIDs)
136
beforeDedup := len(allNewOps)
137
var batchNewOps []plcclient.PLCOperation
138
139
for _, op := range batch {
140
if !seenCIDs[op.CID] {
141
batchNewOps = append(batchNewOps, op)
142
}
143
}
144
145
uniqueInBatch := len(batchNewOps)
146
+
dupesFiltered := afterStripSize - uniqueInBatch
147
+
totalDupes += dupesFiltered + strippedCount
148
149
+
// Try to add to mempool
150
if uniqueInBatch > 0 && mempool != nil {
151
+
_, addErr := mempool.Add(batchNewOps)
152
153
if addErr != nil {
154
+
// Add failed - don't mark as seen
155
if !quiet {
156
f.logger.Printf(" ❌ Mempool add failed: %v", addErr)
157
}
158
mempool.Save()
159
+
return allNewOps, fetchesMade, fmt.Errorf("mempool add failed: %w", addErr)
160
}
161
162
+
// Success - mark as seen
163
for _, op := range batchNewOps {
164
seenCIDs[op.CID] = true
165
}
···
167
168
uniqueAdded := len(allNewOps) - beforeDedup
169
170
if !quiet {
171
opsPerSec := float64(originalBatchSize) / fetchDuration.Seconds()
172
+
if dupesFiltered+strippedCount > 0 {
173
+
f.logger.Printf(" → +%d unique (%d dupes, %d boundary) in %s • Running: %d/%d (%.0f ops/sec)",
174
+
uniqueAdded, dupesFiltered, strippedCount, fetchDuration, len(allNewOps), target, opsPerSec)
175
} else {
176
f.logger.Printf(" → +%d unique in %s • Running: %d/%d (%.0f ops/sec)",
177
uniqueAdded, fetchDuration, len(allNewOps), target, opsPerSec)
178
}
179
}
180
181
+
// ✅ CRITICAL: Calculate NEW boundary CIDs from this fetch for next iteration
182
+
if len(batch) > 0 {
183
+
boundaryTime, newBoundaryCIDs := f.operations.GetBoundaryCIDs(batch)
184
+
currentBoundaryCIDs = newBoundaryCIDs
185
+
currentAfter = boundaryTime.Format(time.RFC3339Nano)
186
+
187
+
if !quiet && len(newBoundaryCIDs) > 1 {
188
+
f.logger.Printf(" Updated boundaries: %d CIDs at %s",
189
+
len(newBoundaryCIDs), currentAfter[:19])
190
+
}
191
+
}
192
+
193
+
// Save if threshold met
194
if err := mempool.SaveIfNeeded(); err != nil {
195
f.logger.Printf(" Warning: failed to save mempool: %v", err)
196
}
197
198
} else if uniqueInBatch > 0 {
199
+
// No mempool - just collect
200
for _, op := range batchNewOps {
201
seenCIDs[op.CID] = true
202
}
203
allNewOps = append(allNewOps, batchNewOps...)
204
205
+
// ✅ Still update boundaries even without mempool
206
+
if len(batch) > 0 {
207
+
boundaryTime, newBoundaryCIDs := f.operations.GetBoundaryCIDs(batch)
208
+
currentBoundaryCIDs = newBoundaryCIDs
209
+
currentAfter = boundaryTime.Format(time.RFC3339Nano)
210
+
}
211
}
212
213
+
// Check if incomplete batch (caught up)
214
if originalBatchSize < batchSize {
215
if !quiet {
216
f.logger.Printf(" Incomplete batch (%d/%d) → caught up", originalBatchSize, batchSize)