Method syncToRemoteWithLock

vfs.go:1900–1967

syncToRemoteWithLock syncs dirty pages to the remote replica. Caller must hold f.mu.

()

// syncToRemoteWithLock syncs dirty pages to the remote replica.
// Caller must hold f.mu.
func (f *VFSFile) syncToRemoteWithLock() error {
	// Double-check dirty pages exist
	if len(f.dirty) == 0 {
		return nil
	}

	ctx := f.ctx

	// Check for conflicts
	if err := f.checkForConflict(ctx); err != nil {
		return err
	}

	// Create LTX file from dirty pages
	ltxReader := f.createLTXFromDirty()

	// Upload LTX file to remote
	info, err := f.client.WriteLTXFile(ctx, 0, f.pendingTXID, f.pendingTXID, ltxReader)
	if err != nil {
		return fmt.Errorf("upload LTX: %w", err)
	}

	f.logger.Info("synced to remote",
		"txid", info.MaxTXID,
		"pages", len(f.dirty),
		"size", info.Size)

	f.expectedTXID = f.pendingTXID
	f.pendingTXID++
	f.pos = ltx.Pos{TXID: f.expectedTXID}

	if f.vfs != nil {
		f.vfs.writeMu.Lock()
		if f.expectedTXID > f.vfs.lastSyncedTXID {
			f.vfs.lastSyncedTXID = f.expectedTXID
		}
		f.vfs.writeMu.Unlock()
	}

	// Update cache with synced pages (index will be populated naturally when pages are fetched)
	for pgno, bufferOff := range f.dirty {
		cachedData := make([]byte, f.pageSize)
		if _, err := f.bufferFile.ReadAt(cachedData, bufferOff); err != nil {
			return fmt.Errorf("read page %d from buffer for cache: %w", pgno, err)
		}
		f.cache.Add(pgno, cachedData)
	}

	// Apply synced pages to hydrated file if hydration is complete
	// Must be done before clearing f.dirty since we need the page offsets
	if f.hydrator != nil && f.hydrator.Complete() {
		if err := f.applySyncedPagesToHydratedFile(); err != nil {
			f.logger.Error("failed to apply synced pages to hydrated file", "error", err)
			// Don't fail the sync - hydration will catch up on next poll
		}
	}

	// Clear dirty pages