Method sync

db.go:1807–2061

sync copies pending bytes from the real WAL to LTX. Returns synced=true if an LTX file was created (i.e., there were new pages to sync).

(ctx context.Context, checkpointing bool, exec *syncExecutor, info syncInfo)

// sync copies pending bytes from the real WAL to LTX.
// Returns synced=true if an LTX file was created (i.e., there were new pages to sync).
func (db *DB) sync(ctx context.Context, checkpointing bool, exec *syncExecutor, info syncInfo) (result syncResult, err error) {
	result.newWALSize = exec.state.lastSyncedWALOffset
	result.syncedToWALEnd = exec.state.syncedToWALEnd
	if info.clearSyncedToWALEnd {
		result.syncedToWALEnd = false
	}

	// Determine the next sequential transaction ID.
	txID := exec.pos.TXID + 1
	db.setSyncDiagPhase(diagPhaseSyncOpenLTX,
		func(s *diagState) {
			s.txID = txID
			s.snapshotting = info.snapshotting
			s.reason = info.reason
		})

	filename := db.LTXPath(0, txID, txID)

	logArgs := []any{
		"txid", txID.String(),
		"offset", info.offset,
	}
	if checkpointing {
		logArgs = append(logArgs, "chkpt", "true")
	}
	if info.snapshotting {
		logArgs = append(logArgs, "snap", "true")
	}
	if info.reason != "" {
		logArgs = append(logArgs, "reason", info.reason)
	}
	db.Logger.Debug("sync", logArgs...)

	// Prevent internal checkpoints during sync. Ignore if already in a checkpoint.
	if !checkpointing {
		db.chkMu.RLock()
		defer db.chkMu.RUnlock()
	}

	fi, err := db.f.Stat()
	if err != nil {
		return result, err
	}
	mode := fi.Mode()
	commit := uint32(fi.Size() / int64(db.pageSize))

	walFile, err := os.Open(db.WALPath())
	if err != nil {
		return result, err
	}
	defer walFile.Close()

	walReaderLogger := db.Logger.With(LogKeySubsystem, LogSubsystemWALReader)
	var rd *WALReader
	if info.offset == WALHeaderSize {
		if rd, err = NewWALReader(walFile, walReaderLogger); err != nil {
			return result, fmt.Errorf("new wal reader: %w", err)
		}