From b9709e92c5d2febbda219b12ecc93e1106d0ae88 Mon Sep 17 00:00:00 2001 From: tree Date: Wed, 12 Nov 2025 07:37:31 +0100 Subject: [PATCH] better index rebuilding --- bundle/manager.go | 148 +++++++++++++++++++------------ cmd/plcbundle/commands/common.go | 4 +- cmd/plcbundle/commands/export.go | 2 +- cmd/plcbundle/commands/sync.go | 2 +- internal/didindex/manager.go | 66 ++++++++++++++ internal/sync/sync_test.go | 41 ++++++--- internal/sync/syncer.go | 24 +++-- 7 files changed, 205 insertions(+), 82 deletions(-) diff --git a/bundle/manager.go b/bundle/manager.go index 8dbb1eb..ba471ef 100644 --- a/bundle/manager.go +++ b/bundle/manager.go @@ -461,7 +461,7 @@ func (m *Manager) loadBundleFromDisk(_ context.Context, bundleNumber int) (*Bund } // SaveBundle saves a bundle to disk and updates the index -func (m *Manager) SaveBundle(ctx context.Context, bundle *Bundle, verbose bool, quiet bool, stats types.BundleProductionStats) (time.Duration, error) { +func (m *Manager) SaveBundle(ctx context.Context, bundle *Bundle, verbose bool, quiet bool, stats types.BundleProductionStats, skipDIDIndex bool) (time.Duration, error) { totalStart := time.Now() if err := bundle.ValidateForSave(); err != nil { @@ -574,7 +574,7 @@ func (m *Manager) SaveBundle(ctx context.Context, bundle *Bundle, verbose bool, // DID index update (if enabled) var indexUpdateDuration time.Duration - if m.didIndex != nil && m.didIndex.Exists() { + if !skipDIDIndex && m.didIndex != nil && m.didIndex.Exists() { indexUpdateStart := time.Now() if err := m.updateDIDIndexForBundle(ctx, bundle); err != nil { m.logger.Printf("Warning: failed to update DID index: %v", err) @@ -1513,14 +1513,13 @@ func (m *Manager) GetMempoolCount() int { return m.mempool.Count() } -// FetchAndSaveNextBundle fetches and saves next bundle, returns bundle number and index time -func (m *Manager) FetchAndSaveNextBundle(ctx context.Context, verbose bool, quiet bool) (int, *types.BundleProductionStats, error) { +func (m *Manager) FetchAndSaveNextBundle(ctx context.Context, verbose bool, quiet bool, skipDIDIndex bool) (int, *types.BundleProductionStats, error) { bundle, stats, err := m.FetchNextBundle(ctx, verbose, quiet) if err != nil { return 0, nil, err } - indexTime, err := m.SaveBundle(ctx, bundle, verbose, quiet, stats) + indexTime, err := m.SaveBundle(ctx, bundle, verbose, quiet, stats, skipDIDIndex) if err != nil { return 0, nil, err } @@ -1536,65 +1535,16 @@ func (m *Manager) RunSyncLoop(ctx context.Context, config *internalsync.SyncLoop } // RunSyncOnce performs a single sync cycle -func (m *Manager) RunSyncOnce(ctx context.Context, config *internalsync.SyncLoopConfig, verbose bool) (int, error) { +func (m *Manager) RunSyncOnce(ctx context.Context, config *internalsync.SyncLoopConfig) (int, error) { // Manager itself implements the SyncManager interface - return internalsync.SyncOnce(ctx, m, config, verbose) + return internalsync.SyncOnce(ctx, m, config) } // EnsureDIDIndex ensures DID index is built and up-to-date // Returns true if index was built/rebuilt, false if already up-to-date func (m *Manager) EnsureDIDIndex(ctx context.Context, progressCallback func(current, total int)) (bool, error) { - bundleCount := m.index.Count() - didStats := m.GetDIDIndexStats() - - if bundleCount == 0 { - return false, nil - } - - needsBuild := false - reason := "" - - if !didStats["exists"].(bool) { - needsBuild = true - reason = "index does not exist" - } else { - // Check version - if m.didIndex != nil { - config := m.didIndex.GetConfig() - if config.Version != didindex.DIDINDEX_VERSION { - needsBuild = true - reason = fmt.Sprintf("index version outdated (v%d, need v%d)", - config.Version, didindex.DIDINDEX_VERSION) - } else { - // Check if index is behind bundles - lastBundle := m.index.GetLastBundle() - if lastBundle != nil && config.LastBundle < lastBundle.BundleNumber { - needsBuild = true - reason = fmt.Sprintf("index is behind (bundle %d, need %d)", - config.LastBundle, lastBundle.BundleNumber) - } - } - } - } - - if !needsBuild { - return false, nil - } - // Build index - m.logger.Printf("Building DID index (%s)", reason) - m.logger.Printf("This may take several minutes...") - - if err := m.BuildDIDIndex(ctx, progressCallback); err != nil { - return false, fmt.Errorf("failed to build DID index: %w", err) - } - - // Verify index consistency - m.logger.Printf("Verifying index consistency...") - if err := m.didIndex.VerifyAndRepairIndex(ctx, m); err != nil { - return false, fmt.Errorf("index verification/repair failed: %w", err) - } - + m.UpdateDIDIndexSmart(ctx, progressCallback) return true, nil } @@ -1843,3 +1793,87 @@ func (m *Manager) ResetResolverStats() { func (m *Manager) SetQuiet(quiet bool) { m.config.Quiet = quiet } + +// ShouldRebuildDIDIndex checks if DID index needs rebuilding +// Returns: (needsRebuild bool, reason string, canUpdateIncrementally bool) +func (m *Manager) ShouldRebuildDIDIndex() (bool, string, bool) { + if m.didIndex == nil { + return false, "DID index disabled", false + } + + needsRebuild, reason := m.didIndex.NeedsRebuild(m.GetBundleIndex()) + + if needsRebuild { + return true, reason, false + } + + // Check if incremental update is better + canIncremental, behindBy := m.didIndex.ShouldUpdateIncrementally(m.GetBundleIndex()) + if canIncremental { + return false, fmt.Sprintf("can update incrementally (%d bundles)", behindBy), true + } + + return false, "index is up to date", false +} + +// UpdateDIDIndexSmart updates DID index intelligently (rebuild vs incremental) +func (m *Manager) UpdateDIDIndexSmart(ctx context.Context, progressCallback func(current, total int)) error { + needsRebuild, reason, canIncremental := m.ShouldRebuildDIDIndex() + + if !needsRebuild && !canIncremental { + if m.config.Verbose { + m.logger.Printf("DID index is up to date") + } + return nil + } + + if needsRebuild { + m.logger.Printf("Rebuilding DID index: %s", reason) + return m.BuildDIDIndex(ctx, progressCallback) + } + + if canIncremental { + m.logger.Printf("Updating DID index incrementally: %s", reason) + return m.updateDIDIndexIncremental(ctx, progressCallback) + } + + return nil +} + +// updateDIDIndexIncremental updates index for missing bundles only +func (m *Manager) updateDIDIndexIncremental(ctx context.Context, progressCallback func(current, total int)) error { + config := m.didIndex.GetConfig() + lastBundle := m.index.GetLastBundle() + + if lastBundle == nil || config.LastBundle >= lastBundle.BundleNumber { + return nil + } + + start := config.LastBundle + 1 + end := lastBundle.BundleNumber + total := end - start + 1 + + m.logger.Printf("Updating DID index for bundles %d-%d (%d bundles)", start, end, total) + + for bundleNum := start; bundleNum <= end; bundleNum++ { + bundle, err := m.LoadBundle(ctx, bundleNum) + if err != nil { + return fmt.Errorf("failed to load bundle %d: %w", bundleNum, err) + } + + bundleData := &didindex.BundleData{ + BundleNumber: bundle.BundleNumber, + Operations: bundle.Operations, + } + + if err := m.didIndex.UpdateIndexForBundle(ctx, bundleData); err != nil { + return fmt.Errorf("failed to update bundle %d: %w", bundleNum, err) + } + + if progressCallback != nil { + progressCallback(bundleNum-start+1, total) + } + } + + return nil +} diff --git a/cmd/plcbundle/commands/common.go b/cmd/plcbundle/commands/common.go index f767d9f..769b172 100644 --- a/cmd/plcbundle/commands/common.go +++ b/cmd/plcbundle/commands/common.go @@ -31,7 +31,7 @@ type BundleManager interface { RefreshMempool() error ClearMempool() error FetchNextBundle(ctx context.Context, verbose bool, quiet bool) (*bundle.Bundle, types.BundleProductionStats, error) - SaveBundle(ctx context.Context, b *bundle.Bundle, verbose bool, quiet bool, stats types.BundleProductionStats) (time.Duration, error) + SaveBundle(ctx context.Context, bundle *bundle.Bundle, verbose bool, quiet bool, stats types.BundleProductionStats, skipDIDIndex bool) (time.Duration, error) SaveIndex() error GetDIDIndexStats() map[string]interface{} GetDIDIndex() *didindex.Manager @@ -44,7 +44,7 @@ type BundleManager interface { LoadOperations(ctx context.Context, bundleNumber int, positions []int) (map[int]*plcclient.PLCOperation, error) CloneFromRemote(ctx context.Context, opts internalsync.CloneOptions) (*internalsync.CloneResult, error) ResolveDID(ctx context.Context, did string) (*bundle.ResolveDIDResult, error) - RunSyncOnce(ctx context.Context, config *internalsync.SyncLoopConfig, verbose bool) (int, error) + RunSyncOnce(ctx context.Context, config *internalsync.SyncLoopConfig) (int, error) RunSyncLoop(ctx context.Context, config *internalsync.SyncLoopConfig) error GetBundleIndex() didindex.BundleIndexProvider ScanDirectoryParallel(workers int, progressCallback func(current, total int, bytesProcessed int64)) (*bundle.DirectoryScanResult, error) diff --git a/cmd/plcbundle/commands/export.go b/cmd/plcbundle/commands/export.go index e5b845f..a1db4dd 100644 --- a/cmd/plcbundle/commands/export.go +++ b/cmd/plcbundle/commands/export.go @@ -238,7 +238,7 @@ func exportBundles(ctx context.Context, mgr BundleManager, opts exportOptions) e } var err error - fetchedCount, err = mgr.RunSyncOnce(ctx, config, opts.verbose) + fetchedCount, err = mgr.RunSyncOnce(ctx, config) if err != nil { return err } diff --git a/cmd/plcbundle/commands/sync.go b/cmd/plcbundle/commands/sync.go index 70d0d5d..1d2670a 100644 --- a/cmd/plcbundle/commands/sync.go +++ b/cmd/plcbundle/commands/sync.go @@ -112,7 +112,7 @@ func runSingleSync(ctx context.Context, mgr *bundle.Manager, maxBundles int, ver } // Call manager method (not internal directly) - synced, err := mgr.RunSyncOnce(ctx, config, verbose) + synced, err := mgr.RunSyncOnce(ctx, config) if err != nil { return err } diff --git a/internal/didindex/manager.go b/internal/didindex/manager.go index e8ecfbe..dee4b53 100644 --- a/internal/didindex/manager.go +++ b/internal/didindex/manager.go @@ -999,3 +999,69 @@ func (dim *Manager) ResetPerformanceStats() { dim.recentLookupIdx = 0 dim.lookupTimeLock.Unlock() } + +// NeedsRebuild checks if index needs rebuilding and returns reason +func (dim *Manager) NeedsRebuild(bundleProvider BundleIndexProvider) (bool, string) { + // Check if index exists + if !dim.Exists() { + return true, "index does not exist" + } + + // Get repository state + bundles := bundleProvider.GetBundles() + if len(bundles) == 0 { + return false, "" // No bundles, no need to rebuild + } + + lastBundleInRepo := bundles[len(bundles)-1].BundleNumber + + // Check version + if dim.config.Version != DIDINDEX_VERSION { + return true, fmt.Sprintf("index version outdated (v%d, need v%d)", + dim.config.Version, DIDINDEX_VERSION) + } + + // Check if index is behind + if dim.config.LastBundle < lastBundleInRepo { + bundlesBehind := lastBundleInRepo - dim.config.LastBundle + + // Smart logic: only rebuild if significantly behind + // Otherwise can do incremental update + if bundlesBehind > 100 { + return true, fmt.Sprintf("index significantly behind (%d bundles)", bundlesBehind) + } + + return false, fmt.Sprintf("index slightly behind (%d bundles) - can update incrementally", bundlesBehind) + } + + // Check if index is ahead (corruption indicator) + if dim.config.LastBundle > lastBundleInRepo { + return true, fmt.Sprintf("index is ahead of repository (has %d, repo has %d) - likely corrupted", + dim.config.LastBundle, lastBundleInRepo) + } + + // Index is up to date + return false, "" +} + +// ShouldUpdateIncrementally checks if incremental update is appropriate +func (dim *Manager) ShouldUpdateIncrementally(bundleProvider BundleIndexProvider) (bool, int) { + if !dim.Exists() { + return false, 0 + } + + bundles := bundleProvider.GetBundles() + if len(bundles) == 0 { + return false, 0 + } + + lastBundleInRepo := bundles[len(bundles)-1].BundleNumber + bundlesBehind := lastBundleInRepo - dim.config.LastBundle + + // Only do incremental if behind by less than 100 bundles + if bundlesBehind > 0 && bundlesBehind <= 100 { + return true, bundlesBehind + } + + return false, 0 +} diff --git a/internal/sync/sync_test.go b/internal/sync/sync_test.go index fd27f7e..3cb5340 100644 --- a/internal/sync/sync_test.go +++ b/internal/sync/sync_test.go @@ -458,13 +458,14 @@ func TestSyncLoopBehavior(t *testing.T) { logger := &testLogger{t: t} config := &internalsync.SyncLoopConfig{ - MaxBundles: 0, - Verbose: false, - Logger: logger, + MaxBundles: 0, + Verbose: false, + Logger: logger, + SkipDIDIndex: false, } // First sync should detect "caught up" when no progress - synced, err := internalsync.SyncOnce(context.Background(), mockMgr, config, false) + synced, err := internalsync.SyncOnce(context.Background(), mockMgr, config) if err != nil { t.Fatalf("SyncOnce failed: %v", err) @@ -484,13 +485,14 @@ func TestSyncLoopBehavior(t *testing.T) { logger := &testLogger{t: t} config := &internalsync.SyncLoopConfig{ - MaxBundles: 3, - Verbose: false, - Logger: logger, + MaxBundles: 3, + Verbose: false, + Logger: logger, + SkipDIDIndex: false, } ctx := context.Background() - synced, err := internalsync.SyncOnce(ctx, mockMgr, config, false) + synced, err := internalsync.SyncOnce(ctx, mockMgr, config) if err != nil { t.Fatalf("SyncOnce failed: %v", err) @@ -511,10 +513,11 @@ func TestSyncLoopBehavior(t *testing.T) { logger := &testLogger{t: t} config := &internalsync.SyncLoopConfig{ - Interval: 100 * time.Millisecond, - MaxBundles: 0, - Verbose: false, - Logger: logger, + Interval: 100 * time.Millisecond, + MaxBundles: 0, + Verbose: false, + Logger: logger, + SkipDIDIndex: false, } ctx, cancel := context.WithCancel(context.Background()) @@ -684,13 +687,25 @@ func (m *mockSyncManager) GetLastBundleNumber() int { return m.lastBundle } +func (m *mockSyncManager) UpdateDIDIndexSmart(ctx context.Context, progressCallback func(current, total int)) error { + m.mu.Lock() + defer m.mu.Unlock() + return nil +} + +func (m *mockSyncManager) BuildDIDIndex(ctx context.Context, progressCallback func(current, total int)) error { + m.mu.Lock() + defer m.mu.Unlock() + return nil +} + func (m *mockSyncManager) GetMempoolCount() int { m.mu.Lock() defer m.mu.Unlock() return m.mempoolCount } -func (m *mockSyncManager) FetchAndSaveNextBundle(ctx context.Context, verbose bool, quiet bool) (int, *types.BundleProductionStats, error) { +func (m *mockSyncManager) FetchAndSaveNextBundle(ctx context.Context, verbose bool, quiet bool, skipDIDIndex bool) (int, *types.BundleProductionStats, error) { m.mu.Lock() defer m.mu.Unlock() diff --git a/internal/sync/syncer.go b/internal/sync/syncer.go index f9862d7..7ad1e1f 100644 --- a/internal/sync/syncer.go +++ b/internal/sync/syncer.go @@ -15,14 +15,18 @@ type SyncLoopConfig struct { Verbose bool Logger types.Logger OnBundleSynced func(bundleNum int, fetchedCount int, mempoolCount int, duration time.Duration, indexTime time.Duration) + SkipDIDIndex bool + Quiet bool } // DefaultSyncLoopConfig returns default configuration func DefaultSyncLoopConfig() *SyncLoopConfig { return &SyncLoopConfig{ - Interval: 1 * time.Minute, - MaxBundles: 0, - Verbose: false, + Interval: 1 * time.Minute, + MaxBundles: 0, + Verbose: false, + SkipDIDIndex: false, + Quiet: false, } } @@ -31,12 +35,14 @@ type SyncManager interface { GetLastBundleNumber() int GetMempoolCount() int // Returns: bundleNumber, indexUpdateTime, error - FetchAndSaveNextBundle(ctx context.Context, verbose bool, quiet bool) (int, *types.BundleProductionStats, error) + FetchAndSaveNextBundle(ctx context.Context, verbose bool, quiet bool, skipDIDIndex bool) (int, *types.BundleProductionStats, error) SaveMempool() error + BuildDIDIndex(ctx context.Context, progressCallback func(current, total int)) error + UpdateDIDIndexSmart(ctx context.Context, progressCallback func(current, total int)) error } // SyncOnce performs a single sync cycle - fetches until caught up -func SyncOnce(ctx context.Context, mgr SyncManager, config *SyncLoopConfig, verbose bool) (int, error) { +func SyncOnce(ctx context.Context, mgr SyncManager, config *SyncLoopConfig) (int, error) { cycleStart := time.Now() startMempool := mgr.GetMempoolCount() @@ -50,7 +56,7 @@ func SyncOnce(ctx context.Context, mgr SyncManager, config *SyncLoopConfig, verb mempoolBefore := mgr.GetMempoolCount() // Attempt to fetch and save next bundle - bundleNum, stats, err := mgr.FetchAndSaveNextBundle(ctx, verbose, false) + bundleNum, stats, err := mgr.FetchAndSaveNextBundle(ctx, config.Verbose, config.Quiet, config.SkipDIDIndex) // Check if we made any progress bundleAfter := mgr.GetLastBundleNumber() @@ -136,11 +142,13 @@ func RunSyncLoop(ctx context.Context, mgr SyncManager, config *SyncLoopConfig) e config.Logger.Printf("[Sync] Initial sync starting...") } - synced, err := SyncOnce(ctx, mgr, config, config.Verbose) + config.SkipDIDIndex = true + synced, err := SyncOnce(ctx, mgr, config) if err != nil { return err } bundlesSynced += synced + mgr.UpdateDIDIndexSmart(ctx, nil) // Check if reached limit if config.MaxBundles > 0 && bundlesSynced >= config.MaxBundles { @@ -167,7 +175,7 @@ func RunSyncLoop(ctx context.Context, mgr SyncManager, config *SyncLoopConfig) e case <-ticker.C: // Each tick, do one sync cycle (which fetches until caught up) - synced, err := SyncOnce(ctx, mgr, config, config.Verbose) + synced, err := SyncOnce(ctx, mgr, config) if err != nil { if config.Logger != nil { config.Logger.Printf("[Sync] Error: %v", err) -- knotmirror 0.0.1