+31 -22

bundle/manager.go

··· 470 470 471 471 // Update DID index if enabled (ONLY when bundle is created) 472 472 if m.didIndex != nil && m.didIndex.Exists() { 473 473 + indexUpdateStart := time.Now() 474 474 + 473 475 if err := m.updateDIDIndexForBundle(ctx, bundle); err != nil { 474 476 m.logger.Printf("Warning: failed to update DID index: %v", err) 477 477 + } else { 478 478 + indexUpdateDuration := time.Since(indexUpdateStart) 479 479 + 480 480 + if !quiet { 481 481 + m.logger.Printf(" [DID Index] Updated in %s", indexUpdateDuration) 482 482 + } 475 483 } 476 484 } 477 485 ··· 1099 1107 } 1100 1108 } 1101 1109 1102 1102 - // ✨ Use mempool's last time if available 1110 1110 + // Use mempool's last time if available 1103 1111 if m.mempool.Count() > 0 { 1104 1112 mempoolLastTime := m.mempool.GetLastTime() 1105 1113 if mempoolLastTime != "" { ··· 1115 1123 m.logger.Printf("Preparing bundle %06d (mempool: %d ops)...", nextBundleNum, m.mempool.Count()) 1116 1124 } 1117 1125 1118 1118 - // ✨ NEW: Loop until we have enough OR catch up to latest 1119 1119 - maxAttempts := 50 // Safety limit 1126 1126 + // Track total fetches across all attempts 1127 1127 + totalFetches := 0 1128 1128 + maxAttempts := 50 1120 1129 attempt := 0 1121 1130 caughtUp := false 1131 1131 + attemptStart := time.Now() 1122 1132 1123 1133 for m.mempool.Count() < types.BUNDLE_SIZE && attempt < maxAttempts { 1124 1134 attempt++ 1125 1125 - 1126 1135 needed := types.BUNDLE_SIZE - m.mempool.Count() 1127 1136 1128 1137 if !quiet && attempt > 1 { 1129 1138 m.logger.Printf(" Attempt %d: Need %d more ops...", attempt, needed) 1130 1139 } 1131 1140 1132 1132 - newOps, err := m.syncer.FetchToMempool( 1141 1141 + newOps, fetchCount, err := m.syncer.FetchToMempool( 1133 1142 ctx, 1134 1143 afterTime, 1135 1144 prevBoundaryCIDs, 1136 1145 needed, 1137 1146 quiet, 1138 1147 m.mempool.Count(), 1148 1148 + totalFetches, // Pass current total 1139 1149 ) 1140 1150 1151 1151 + // Update total fetch counter 1152 1152 + totalFetches += fetchCount 1153 1153 + 1154 1154 + // Check if we got an incomplete batch 1155 1155 + gotIncompleteBatch := len(newOps) > 0 && len(newOps) < needed && err == nil 1156 1156 + 1141 1157 // Add operations if we got any 1142 1158 if len(newOps) > 0 { 1143 1159 added, addErr := m.mempool.Add(newOps) ··· 1151 1167 } 1152 1168 1153 1169 // Update cursor for next fetch 1154 1154 - if len(newOps) > 0 { 1155 1155 - afterTime = newOps[len(newOps)-1].CreatedAt.Format(time.RFC3339Nano) 1156 1156 - } 1170 1170 + afterTime = newOps[len(newOps)-1].CreatedAt.Format(time.RFC3339Nano) 1157 1171 } 1158 1172 1159 1159 - // Check if we caught up (got incomplete batch) 1160 1160 - if err != nil || len(newOps) == 0 { 1173 1173 + // Stop if caught up or error 1174 1174 + if err != nil || len(newOps) == 0 || gotIncompleteBatch { 1161 1175 caughtUp = true 1162 1162 - if !quiet { 1176 1176 + if !quiet && totalFetches > 0 { 1163 1177 m.logger.Printf(" Caught up to latest PLC data") 1164 1178 } 1165 1179 break 1166 1180 } 1167 1181 1168 1168 - // If we got a full batch but still need more, continue looping 1169 1169 - if len(newOps) > 0 && m.mempool.Count() < types.BUNDLE_SIZE { 1170 1170 - if !quiet { 1171 1171 - m.logger.Printf(" Got full batch, fetching more...") 1172 1172 - } 1173 1173 - continue 1174 1174 - } 1175 1175 - 1176 1182 // If we have enough, break 1177 1183 if m.mempool.Count() >= types.BUNDLE_SIZE { 1178 1184 break ··· 1182 1188 // Save mempool state 1183 1189 m.mempool.Save() 1184 1190 1191 1191 + totalDuration := time.Since(attemptStart) 1192 1192 + 1185 1193 // Check if we have enough for a bundle 1186 1194 if m.mempool.Count() < types.BUNDLE_SIZE { 1187 1195 if caughtUp { ··· 1202 1210 // Create bundle structure 1203 1211 syncBundle := internalsync.CreateBundle(nextBundleNum, operations, afterTime, prevBundleHash, m.operations) 1204 1212 1205 1205 - // Convert to bundle.Bundle 1206 1213 bundle := &Bundle{ 1207 1214 BundleNumber: syncBundle.BundleNumber, 1208 1215 StartTime: syncBundle.StartTime, ··· 1217 1224 } 1218 1225 1219 1226 if !quiet { 1220 1220 - m.logger.Printf("✓ Bundle %06d ready (%d ops, %d DIDs)", 1221 1221 - bundle.BundleNumber, len(bundle.Operations), bundle.DIDCount) 1227 1227 + avgPerFetch := float64(types.BUNDLE_SIZE) / float64(totalFetches) 1228 1228 + m.logger.Printf("✓ Bundle %06d ready (%d ops, %d DIDs) - %d fetches in %s (avg %.0f ops/fetch)", 1229 1229 + bundle.BundleNumber, len(bundle.Operations), bundle.DIDCount, 1230 1230 + totalFetches, totalDuration.Round(time.Millisecond), avgPerFetch) 1222 1231 } 1223 1232 1224 1233 return bundle, nil

+109 -14

internal/didindex/builder.go

··· 6 6 "fmt" 7 7 "os" 8 8 "path/filepath" 9 9 + "runtime" 9 10 "sort" 11 11 + "sync" 12 12 + "time" 10 13 ) 11 14 12 15 // newShardBuilder creates a new shard builder ··· 214 217 return int64(len(builder.entries)), nil 215 218 } 216 219 217 217 - // UpdateIndexForBundle adds operations from a new bundle (incremental + ATOMIC) 220 220 + // UpdateIndexForBundle adds operations from a new bundle (incremental + ATOMIC + PARALLEL) 218 221 func (dim *Manager) UpdateIndexForBundle(ctx context.Context, bundle *BundleData) error { 219 222 dim.indexMu.Lock() 220 223 defer dim.indexMu.Unlock() 221 224 222 222 - // Group operations by shard 225 225 + totalStart := time.Now() 226 226 + 227 227 + // STEP 1: Group operations by shard 228 228 + groupStart := time.Now() 223 229 shardOps := make(map[uint8]map[string][]OpLocation) 224 230 225 231 for pos, op := range bundle.Operations { ··· 241 247 }) 242 248 } 243 249 244 244 - // PHASE 1: Write ALL shards to .tmp files FIRST 250 250 + groupDuration := time.Since(groupStart) 251 251 + dim.logger.Printf(" [DID Index] Grouped operations into %d shards in %s", 252 252 + len(shardOps), groupDuration) 253 253 + 254 254 + // STEP 2: Write ALL shards to .tmp files FIRST (PARALLEL) 255 255 + writeStart := time.Now() 256 256 + 245 257 tmpShards := make(map[uint8]string) 258 258 + var tmpShardsMu sync.Mutex 246 259 var deltaCount int64 260 260 + var deltaCountMu sync.Mutex 247 261 262 262 + // Error handling 263 263 + errChan := make(chan error, len(shardOps)) 264 264 + 265 265 + // Worker pool 266 266 + workers := runtime.NumCPU() 267 267 + if workers > len(shardOps) { 268 268 + workers = len(shardOps) 269 269 + } 270 270 + if workers < 1 { 271 271 + workers = 1 272 272 + } 273 273 + 274 274 + semaphore := make(chan struct{}, workers) 275 275 + var wg sync.WaitGroup 276 276 + 277 277 + dim.logger.Printf(" [DID Index] Updating %d shards in parallel (%d workers)...", 278 278 + len(shardOps), workers) 279 279 + 280 280 + // Process each shard in parallel 248 281 for shardNum, newOps := range shardOps { 249 249 - tmpPath := filepath.Join(dim.shardDir, fmt.Sprintf("%02x.idx.tmp", shardNum)) 282 282 + wg.Add(1) 283 283 + go func(sNum uint8, ops map[string][]OpLocation) { 284 284 + defer wg.Done() 250 285 251 251 - addedCount, err := dim.updateShardToTemp(shardNum, newOps, tmpPath) 252 252 - if err != nil { 253 253 - dim.cleanupTempShards(tmpShards) 254 254 - return fmt.Errorf("failed to prepare shard %02x: %w", shardNum, err) 255 255 - } 286 286 + // Acquire semaphore (limit concurrency) 287 287 + semaphore <- struct{}{} 288 288 + defer func() { <-semaphore }() 256 289 257 257 - tmpShards[shardNum] = tmpPath 258 258 - deltaCount += addedCount 290 290 + shardStart := time.Now() 291 291 + tmpPath := filepath.Join(dim.shardDir, fmt.Sprintf("%02x.idx.tmp", sNum)) 292 292 + 293 293 + addedCount, err := dim.updateShardToTemp(sNum, ops, tmpPath) 294 294 + if err != nil { 295 295 + errChan <- fmt.Errorf("shard %02x: %w", sNum, err) 296 296 + return 297 297 + } 298 298 + 299 299 + shardDuration := time.Since(shardStart) 300 300 + 301 301 + // Update shared state 302 302 + tmpShardsMu.Lock() 303 303 + tmpShards[sNum] = tmpPath 304 304 + tmpShardsMu.Unlock() 305 305 + 306 306 + deltaCountMu.Lock() 307 307 + deltaCount += addedCount 308 308 + deltaCountMu.Unlock() 309 309 + 310 310 + // Debug log for each shard 311 311 + if dim.verbose { 312 312 + dim.logger.Printf(" Shard %02x: +%d DIDs in %s (%d ops)", 313 313 + sNum, addedCount, shardDuration, len(ops)) 314 314 + } 315 315 + }(shardNum, newOps) 259 316 } 260 317 261 261 - // PHASE 2: Atomically commit ALL shards 318 318 + // Wait for all workers 319 319 + wg.Wait() 320 320 + close(errChan) 321 321 + 322 322 + writeDuration := time.Since(writeStart) 323 323 + dim.logger.Printf(" [DID Index] Wrote %d temp files in %s (%.1f shards/sec)", 324 324 + len(tmpShards), writeDuration, float64(len(tmpShards))/writeDuration.Seconds()) 325 325 + 326 326 + // Check for errors 327 327 + if err := <-errChan; err != nil { 328 328 + dim.cleanupTempShards(tmpShards) 329 329 + return err 330 330 + } 331 331 + 332 332 + // STEP 3: Atomically commit ALL shards 333 333 + commitStart := time.Now() 334 334 + 262 335 for shardNum, tmpPath := range tmpShards { 263 336 finalPath := filepath.Join(dim.shardDir, fmt.Sprintf("%02x.idx", shardNum)) 264 337 ··· 271 344 dim.invalidateShard(shardNum) 272 345 } 273 346 274 274 - // PHASE 3: Update config 347 347 + commitDuration := time.Since(commitStart) 348 348 + 349 349 + // STEP 4: Update config 350 350 + configStart := time.Now() 351 351 + 275 352 dim.config.TotalDIDs += deltaCount 276 353 dim.config.LastBundle = bundle.BundleNumber 277 354 278 278 - return dim.saveIndexConfig() 355 355 + if err := dim.saveIndexConfig(); err != nil { 356 356 + return fmt.Errorf("failed to save config: %w", err) 357 357 + } 358 358 + 359 359 + configDuration := time.Since(configStart) 360 360 + totalDuration := time.Since(totalStart) 361 361 + 362 362 + // Summary log 363 363 + dim.logger.Printf(" [DID Index] ✓ Bundle %06d indexed: +%d DIDs, %d shards updated in %s", 364 364 + bundle.BundleNumber, deltaCount, len(tmpShards), totalDuration) 365 365 + 366 366 + if dim.verbose { 367 367 + dim.logger.Printf(" Breakdown: group=%s write=%s commit=%s config=%s", 368 368 + groupDuration, writeDuration, commitDuration, configDuration) 369 369 + dim.logger.Printf(" Throughput: %.0f ops/sec", 370 370 + float64(len(bundle.Operations))/totalDuration.Seconds()) 371 371 + } 372 372 + 373 373 + return nil 279 374 } 280 375 281 376 // updateShardToTemp updates a shard and writes to temp file

+42 -15

internal/sync/fetcher.go

··· 35 35 target int, 36 36 quiet bool, 37 37 currentMempoolCount int, 38 38 - ) ([]plcclient.PLCOperation, error) { 38 38 + totalFetchesSoFar int, 39 39 + ) ([]plcclient.PLCOperation, int, error) { 39 40 40 41 seenCIDs := make(map[string]bool) 41 42 ··· 47 48 currentAfter := afterTime 48 49 maxFetches := 20 49 50 var allNewOps []plcclient.PLCOperation 51 51 + fetchesMade := 0 50 52 51 53 for fetchNum := 0; fetchNum < maxFetches; fetchNum++ { 54 54 + fetchesMade++ 52 55 remaining := target - len(allNewOps) 53 56 if remaining <= 0 { 54 57 break 55 58 } 56 59 57 57 - batchSize := 1000 58 58 - if remaining < 500 { 60 60 + // ✨ SMART BATCH SIZING 61 61 + var batchSize int 62 62 + switch { 63 63 + case remaining <= 50: 64 64 + batchSize = 50 // Fetch exactly what we need (with small buffer) 65 65 + case remaining <= 100: 66 66 + batchSize = 100 67 67 + case remaining <= 500: 59 68 batchSize = 200 69 69 + default: 70 70 + batchSize = 1000 60 71 } 72 72 + 73 73 + fetchStart := time.Now() 61 74 62 75 if !quiet { 63 63 - f.logger.Printf(" Fetch #%d: requesting %d operations (after: %s)", 64 64 - fetchNum+1, batchSize, currentAfter[:19]) 76 76 + f.logger.Printf(" Fetch #%d: requesting %d operations (need %d, after: %s)", 77 77 + totalFetchesSoFar+fetchesMade, batchSize, remaining, currentAfter[:19]) 65 78 } 66 79 67 80 batch, err := f.plcClient.Export(ctx, plcclient.ExportOptions{ ··· 69 82 After: currentAfter, 70 83 }) 71 84 85 85 + fetchDuration := time.Since(fetchStart) 86 86 + 72 87 if err != nil { 73 73 - return allNewOps, fmt.Errorf("export failed: %w", err) 88 88 + return allNewOps, fetchesMade, fmt.Errorf("export failed: %w", err) 74 89 } 75 90 76 91 if len(batch) == 0 { 77 92 if !quiet { 78 78 - f.logger.Printf(" No more operations available from PLC") 93 93 + f.logger.Printf(" No more operations available from PLC (in %s)", fetchDuration) 79 94 } 80 80 - // Return what we have (might be incomplete) 81 81 - return allNewOps, nil 95 95 + return allNewOps, fetchesMade, nil 82 96 } 83 97 84 98 // Deduplicate ··· 90 104 } 91 105 } 92 106 93 93 - if !quiet && len(batch) > 0 { 94 94 - deduped := len(batch) - (len(allNewOps) - beforeDedup) 107 107 + uniqueAdded := len(allNewOps) - beforeDedup 108 108 + deduped := len(batch) - uniqueAdded 109 109 + 110 110 + // ✨ DETAILED METRICS 111 111 + if !quiet { 112 112 + opsPerSec := float64(len(batch)) / fetchDuration.Seconds() 113 113 + 95 114 if deduped > 0 { 96 96 - f.logger.Printf(" Received %d ops (%d duplicates filtered)", len(batch), deduped) 115 115 + f.logger.Printf(" Received %d ops (%d unique, %d dupes) in %s (%.0f ops/sec)", 116 116 + len(batch), uniqueAdded, deduped, fetchDuration, opsPerSec) 117 117 + } else { 118 118 + f.logger.Printf(" Received %d ops in %s (%.0f ops/sec)", 119 119 + len(batch), fetchDuration, opsPerSec) 97 120 } 98 121 } 99 122 ··· 102 125 currentAfter = batch[len(batch)-1].CreatedAt.Format(time.RFC3339Nano) 103 126 } 104 127 105 105 - // ✨ KEY: Stop if we got incomplete batch (caught up!) 128 128 + // Stop if we got incomplete batch (caught up!) 106 129 if len(batch) < batchSize { 107 130 if !quiet { 108 131 f.logger.Printf(" Received incomplete batch (%d/%d) → caught up to latest", 109 132 len(batch), batchSize) 110 133 } 111 111 - return allNewOps, nil 134 134 + return allNewOps, fetchesMade, nil 112 135 } 113 136 114 137 // If we have enough, stop 115 138 if len(allNewOps) >= target { 139 139 + if !quiet { 140 140 + f.logger.Printf(" ✓ Target reached (%d/%d unique ops collected)", 141 141 + len(allNewOps), target) 142 142 + } 116 143 break 117 144 } 118 145 } 119 146 120 120 - return allNewOps, nil 147 147 + return allNewOps, fetchesMade, nil 121 148 }