feat: fill the gap between bulk & live (#696)

Author: chris-4chain

pkg/services/chaintracks/bulk_headers_container.go

@@ -250,3 +250,41 @@ func (b *bulkHeadersContainer) GetFileDataByIndex(fileID int) (*ingest.BulkFileD
 
 	return &b.GeneratedFileData[fileID], nil
 }
+
+func (b *bulkHeadersContainer) LastHeader() (*wdk.ChainBlockHeader, *internal.ChainWork, error) {
+	length := len(b.chunks)
+	if length == 0 {
+		return nil, nil, fmt.Errorf("no chunks available to retrieve last header")
+	}
+
+	lastChunk := b.chunks[length-1]
+	headerCount := len(lastChunk.data) / 80
+	if headerCount == 0 {
+		return nil, nil, fmt.Errorf("last chunk contains no headers")
+	}
+
+	headerDataStart := (headerCount - 1) * 80
+	headerDataEnd := headerDataStart + 80
+
+	headerData := lastChunk.data[headerDataStart:headerDataEnd]
+
+	baseBlockHeader, err := wdk.ChainBaseBlockHeaderFromBytes(headerData)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to parse last block header: %w", err)
+	}
+
+	blockHash, err := baseBlockHeader.CalculateHash()
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to compute hash for last block header: %w", err)
+	}
+
+	height := b.MaxHeightAtChunk(length - 1)
+
+	header := &wdk.ChainBlockHeader{
+		ChainBaseBlockHeader: *baseBlockHeader,
+		Height:               must.ConvertToUInt(height),
+		Hash:                 blockHash.String(),
+	}
+
+	return header, &lastChunk.LastChainWork, nil
+}

pkg/services/chaintracks/bulk_manager.go

@@ -9,6 +9,7 @@ import (
 	"github.com/bsv-blockchain/go-wallet-toolbox/pkg/defs"
 	"github.com/bsv-blockchain/go-wallet-toolbox/pkg/internal/logging"
 	"github.com/bsv-blockchain/go-wallet-toolbox/pkg/services/chaintracks/ingest"
+	"github.com/bsv-blockchain/go-wallet-toolbox/pkg/services/chaintracks/internal"
 	"github.com/bsv-blockchain/go-wallet-toolbox/pkg/services/chaintracks/models"
 	"github.com/bsv-blockchain/go-wallet-toolbox/pkg/wdk"
 )
@@ -33,9 +34,18 @@ func newBulkManager(logger *slog.Logger, bulkIngestors []NamedBulkIngestor, chai
 }
 
 func (bm *bulkManager) SyncBulkStorage(ctx context.Context, presentHeight uint, initialRanges models.HeightRanges) error {
+	if presentHeight <= liveHeightThreshold {
+		bm.logger.Info("Skipping bulk synchronization - present height below live height threshold", slog.Any("present_height", presentHeight), slog.Any("live_height_threshold", liveHeightThreshold))
+		return nil
+	}
+
 	bm.logger.Info("Starting bulk synchronization", slog.Any("present_height", presentHeight), slog.Any("initial_ranges", initialRanges))
 
-	missingRange := models.NewHeightRange(0, presentHeight)
+	missingRange, err := models.NewHeightRange(0, presentHeight-liveHeightThreshold).Subtract(initialRanges.Bulk)
+	if err != nil {
+		return fmt.Errorf("failed to compute missing bulk range: %w", err)
+	}
+
 	for _, ingestor := range bm.bulkIngestors {
 		if missingRange.IsEmpty() {
 			break
@@ -47,7 +57,7 @@ func (bm *bulkManager) SyncBulkStorage(ctx context.Context, presentHeight uint,
 			return fmt.Errorf("bulk synchronization failed for ingestor %s: %w", ingestor.Name, err)
 		}
 
-		if err := bm.processBulkChunks(ctx, bulkChunks, downloader); err != nil {
+		if err := bm.processBulkChunks(ctx, bulkChunks, downloader, missingRange.MaxHeight); err != nil {
 			return fmt.Errorf("failed to process bulk chunks from ingestor %s: %w", ingestor.Name, err)
 		}
 
@@ -80,6 +90,13 @@ func (bm *bulkManager) FindHeaderForHeight(height uint) (*wdk.ChainBlockHeader,
 	return bm.container.FindHeaderForHeight(height)
 }
 
+func (bm *bulkManager) LastHeader() (*wdk.ChainBlockHeader, *internal.ChainWork, error) {
+	bm.locker.RLock()
+	defer bm.locker.RUnlock()
+
+	return bm.container.LastHeader()
+}
+
 func (bm *bulkManager) FilesInfo() *ingest.BulkHeaderFilesInfo {
 	bm.locker.RLock()
 	defer bm.locker.RUnlock()
@@ -97,7 +114,7 @@ func (bm *bulkManager) GetFileDataByIndex(fileID int) (*ingest.BulkFileData, err
 	return bm.container.GetFileDataByIndex(fileID)
 }
 
-func (bm *bulkManager) processBulkChunks(ctx context.Context, bulkChunks []ingest.BulkHeaderMinimumInfo, downloader ingest.BulkFileDownloader) error {
+func (bm *bulkManager) processBulkChunks(ctx context.Context, bulkChunks []ingest.BulkHeaderMinimumInfo, downloader ingest.BulkFileDownloader, maxHeight uint) error {
 	chunksToLoad := bm.getChunksToLoad(bulkChunks)
 	type chunkWithInfo struct {
 		data []byte
@@ -130,7 +147,12 @@ func (bm *bulkManager) processBulkChunks(ctx context.Context, bulkChunks []inges
 			continue
 		}
 
-		if err := bm.container.Add(ctx, fileData.data, fileData.info.ToHeightRange()); err != nil {
+		dataRange := fileData.info.ToHeightRange()
+		if dataRange.MaxHeight > maxHeight {
+			dataRange.MaxHeight = maxHeight
+		}
+
+		if err := bm.container.Add(ctx, fileData.data, dataRange); err != nil {
 			return fmt.Errorf("failed to add bulk file %v to container: %w", fileData.info, err)
 		}
 	}
@@ -157,3 +179,82 @@ func (bm *bulkManager) shouldAddNewFile(info *ingest.BulkHeaderMinimumInfo) bool
 	rangeToAdd := info.ToHeightRange().Above(currentRange)
 	return !rangeToAdd.IsEmpty()
 }
+
+func (bm *bulkManager) GetGapHeadersAsLive(ctx context.Context, presentHeight uint, liveInitialRange models.HeightRange) ([]wdk.ChainBlockHeader, error) {
+	var newLiveHeaders []wdk.ChainBlockHeader
+	maxBulkHeight := bm.GetHeightRange().MaxHeight
+	minLiveHeight := presentHeight
+	if liveInitialRange.NotEmpty() {
+		minLiveHeight = liveInitialRange.MinHeight
+	}
+
+	if minLiveHeight <= maxBulkHeight || minLiveHeight < maxBulkHeight+addLiveRecursionLimit {
+		// no gap to fill
+		return nil, nil
+	}
+
+	// use bulk ingestors to fill the gap and treat fetched headers as live headers
+	missingRange := models.NewHeightRange(maxBulkHeight+1, minLiveHeight-1)
+
+	for _, ingestor := range bm.bulkIngestors {
+		if missingRange.IsEmpty() {
+			break
+		}
+
+		bulkChunks, downloader, err := ingestor.Ingestor.Synchronize(ctx, presentHeight, missingRange)
+		if err != nil {
+			bm.logger.Error("Chaintracks service - error during bulk synchronization to fill gap", slog.String("ingestor_name", ingestor.Name), slog.String("error", err.Error()))
+			return nil, fmt.Errorf("bulk synchronization to fill gap failed for ingestor %s: %w", ingestor.Name, err)
+		}
+
+		for _, chunk := range bulkChunks {
+			intersection := chunk.ToHeightRange().Intersect(missingRange)
+			if intersection.IsEmpty() {
+				continue
+			}
+
+			data, err := downloader(ctx, chunk)
+			if err != nil {
+				return nil, fmt.Errorf("failed to download bulk file %v to fill gap: %w", chunk, err)
+			}
+
+			if err := chunk.Validate(data); err != nil {
+				return nil, fmt.Errorf("downloaded bulk file %v to fill gap is invalid: %w", chunk, err)
+			}
+
+			minIndex := intersection.MinHeight - chunk.FirstHeight
+			maxIndex := intersection.MaxHeight - chunk.FirstHeight
+
+			for i := minIndex; i <= maxIndex; i++ {
+				startByte := i * 80
+				endByte := startByte + 80
+				headerData := data[startByte:endByte]
+
+				baseHeader, err := wdk.ChainBaseBlockHeaderFromBytes(headerData)
+				if err != nil {
+					return nil, fmt.Errorf("failed to parse block header at height %d from bulk file %v: %w", chunk.FirstHeight+i, chunk, err)
+				}
+
+				blockHash, err := baseHeader.CalculateHash()
+				if err != nil {
+					return nil, fmt.Errorf("failed to compute hash for block header at height %d from bulk file %v: %w", chunk.FirstHeight+i, chunk, err)
+				}
+
+				header := wdk.ChainBlockHeader{
+					ChainBaseBlockHeader: *baseHeader,
+					Height:               chunk.FirstHeight + i,
+					Hash:                 blockHash.String(),
+				}
+
+				newLiveHeaders = append(newLiveHeaders, header)
+			}
+
+			missingRange, err = missingRange.Subtract(intersection)
+			if err != nil {
+				return nil, fmt.Errorf("failed to compute missing range after filling gap with ingestor %s: %w", ingestor.Name, err)
+			}
+		}
+	}
+
+	return newLiveHeaders, nil
+}

pkg/services/chaintracks/chaintracks_service.go

@@ -20,12 +20,14 @@ import (
 )
 
 const (
+	// TODO: constants below, can be made configurable if needed
 	liveHeadersChanSize    = 1000
 	lastPresentHeightTTL   = 60 * time.Second
 	cdnSyncRepeatDuration  = 24 * time.Hour
 	syncCheckInterval      = 1 * time.Second
 	addLiveRecursionLimit  = 11
 	halfLiveRecursionLimit = addLiveRecursionLimit / 2
+	liveHeightThreshold    = 2000
 )
 
 // Service provides core functionality for the Chaintracks service with logging and configuration support.
@@ -367,6 +369,10 @@ func (s *Service) shiftLiveHeaders(ctx context.Context) error {
 		return fmt.Errorf("bulk synchronization failed during live headers shift: %w", err)
 	}
 
+	if err := s.fillGapLiveHeaders(ctx, presentHeight, before.Live); err != nil {
+		return fmt.Errorf("failed to fill gap live headers during live headers shift: %w", err)
+	}
+
 	if err := s.processHeaders(ctx); err != nil {
 		return fmt.Errorf("failed to process live headers during live headers shift: %w", err)
 	}
@@ -381,6 +387,25 @@ func (s *Service) skipBulkSync(presentHeight uint, ranges models.HeightRanges) b
 	return ranges.Live.NotEmpty() && ranges.Live.MaxHeight >= presentHeight-halfLiveRecursionLimit
 }
 
+func (s *Service) fillGapLiveHeaders(ctx context.Context, presentHeight uint, liveInitialRange models.HeightRange) error {
+	gapHeaders, err := s.bulkMgr.GetGapHeadersAsLive(ctx, presentHeight, liveInitialRange)
+	if err != nil {
+		return fmt.Errorf("failed to get gap headers as live during live headers shift: %w", err)
+	}
+
+	if len(gapHeaders) > 0 {
+		s.logger.Info("Filling gap between bulk and live headers using bulk ingestors", slog.Any("num_gap_headers", len(gapHeaders)))
+	}
+
+	for _, header := range gapHeaders {
+		if err := s.addLiveHeader(ctx, header); err != nil {
+			s.logger.Warn("Chaintracks service - failed to add gap header as live", slog.String("header_hash", header.Hash), slog.String("error", err.Error()))
+		}
+	}
+
+	return nil
+}
+
 func (s *Service) processHeaders(ctx context.Context) error {
 	for {
 		select {
@@ -441,15 +466,32 @@ func (s *Service) addLiveHeaderRecursive(ctx context.Context, header wdk.ChainBl
 var errNoPrev = fmt.Errorf("no previous header found")
 
 func (s *Service) storeLiveHeader(ctx context.Context, header wdk.ChainBlockHeader) (err error) {
-	// TODO: implement header.Validate() method and uncomment validation check
-	//if err := header.Validate(); err != nil {
-	//	return fmt.Errorf("invalid block header: %w", err)
-	//}
+	if header.Height == 0 {
+		return fmt.Errorf("handling genesis block header is not supported here")
+	}
+
+	if err := header.Validate(); err != nil {
+		return fmt.Errorf("invalid block header: %w", err)
+	}
 
 	if IsDirtyHash(header.Hash) {
 		return fmt.Errorf("cannot add block header with dirty hash: %s", header.Hash)
 	}
 
+	lastBulk, lastBulkChainWork, err := s.bulkMgr.LastHeader()
+	if err != nil {
+		return fmt.Errorf("failed to get last bulk header: %w", err)
+	}
+
+	if lastBulk == nil {
+		return fmt.Errorf("no bulk headers available to validate against")
+	}
+
+	if header.Height <= lastBulk.Height {
+		s.logger.Info("Chaintracks service - skipping storage of live header already present in bulk storage", slog.String("header_hash", header.Hash), slog.Any("header_height", header.Height))
+		return nil
+	}
+
 	q := s.storage.Query(ctx)
 	q.Begin()
 	defer func() {
@@ -479,28 +521,26 @@ func (s *Service) storeLiveHeader(ctx context.Context, header wdk.ChainBlockHead
 	}
 
 	if oneBack == nil {
-		s.logger.Debug("Chaintracks service - previous header not found, cannot add header yet", slog.String("header_hash", header.Hash), slog.String("previous_hash", header.PreviousHash))
-
 		if count, err := q.CountLiveHeaders(); err != nil {
 			return fmt.Errorf("failed to count live headers: %w", err)
 		} else if count == 0 {
-			s.logger.Info("Chaintracks service - no live headers present, inserting genesis header", slog.String("header_hash", header.Hash))
-
-			// TODO check if this first-live-block-header matches the last bulk header
-			// TODO: Important: Chainwork from bits should be added to the last ChainWork from the last bulk file
-			headerChainWork := internal.ChainWorkFromBits(header.Bits)
-
-			if err := q.InsertNewLiveHeader(&models.LiveBlockHeader{
-				ChainBlockHeader: header,
-				PreviousHeaderID: nil,
-				ChainWork:        headerChainWork.To64PadHex(),
-				IsActive:         true,
-				IsChainTip:       true,
-			}); err != nil {
-				return fmt.Errorf("failed to insert genesis live header: %w", err)
+			// No live headers yet, check if this header connects directly to last bulk
+			if header.PreviousHash == lastBulk.Hash && header.Height == lastBulk.Height+1 {
+				headerChainWork := internal.ChainWorkFromBits(header.Bits)
+				chainWork := headerChainWork.AddChainWork(*lastBulkChainWork)
+
+				if err := q.InsertNewLiveHeader(&models.LiveBlockHeader{
+					ChainBlockHeader: header,
+					PreviousHeaderID: nil,
+					ChainWork:        chainWork.To64PadHex(),
+					IsChainTip:       true,
+					IsActive:         true,
+				}); err != nil {
+					return fmt.Errorf("failed to insert new live header: %w", err)
+				}
+
+				return nil
 			}
-
-			return nil
 		}
 
 		return errNoPrev

pkg/services/chaintracks/chaintracks_storage.interface.go

@@ -34,7 +34,6 @@ type NamedLiveIngestor struct {
 
 // BulkIngestor defines an interface for bulk synchronization of block headers within specified height ranges.
 // The Synchronize method ingests headers up to the given presentHeight for provided height ranges and returns insertion results.
-// TODO: refine return type from 'any' to a more specific type representing synchronization results.
 type BulkIngestor interface {
 	Synchronize(ctx context.Context, presentHeight uint, rangeToFetch models.HeightRange) ([]ingest.BulkHeaderMinimumInfo, ingest.BulkFileDownloader, error)
 }

pkg/services/chaintracks/ingest/bulk_ingestor_cdn.go

@@ -40,8 +40,6 @@ type BulkFileDownloader = func(ctx context.Context, fileInfo BulkHeaderMinimumIn
 // Synchronize retrieves available bulk header files for the configured BSV network and prepares chunks for ingestion.
 // It validates file metadata, checks network consistency, and returns a list of chunked header information for sync.
 func (b *BulkIngestorCDN) Synchronize(ctx context.Context, presentHeight uint, rangeToFetch models.HeightRange) ([]BulkHeaderMinimumInfo, BulkFileDownloader, error) {
-	// TODO: PresentHeight and ranges are not used in TS implementation, consider using them for optimization
-
 	filesInfo, err := b.reader.FetchBulkHeaderFilesInfo(ctx, b.chain)
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to fetch bulk header files info: %w", err)
@@ -50,6 +48,15 @@ func (b *BulkIngestorCDN) Synchronize(ctx context.Context, presentHeight uint, r
 	bulkInfo := make([]BulkHeaderMinimumInfo, 0, len(filesInfo.Files))
 	for i := range filesInfo.Files {
 		file := &filesInfo.Files[i]
+
+		if !file.ToHeightRange().Overlaps(rangeToFetch) {
+			b.logger.Info("Skipping bulk header file - does not overlap requested range",
+				slog.String("file_name", file.FileName),
+				logging.Number("start_height", file.FirstHeight),
+				logging.Number("end_height", must.ConvertToIntFromUnsigned(file.FirstHeight)+file.Count-1))
+			continue
+		}
+
 		b.logger.Info("Found bulk header file",
 			slog.String("file_name", file.FileName),
 			logging.Number("start_height", file.FirstHeight),