Repository: pingcap/tidb
File: br/pkg/task/stream.go (Go)

Code before:
	shiftPhysical := physical - streamShiftDuration.Milliseconds()
	if shiftPhysical < 0 {
		return 0
	}
	return oracle.ComposeTS(shiftPhysical, logical)
}

func buildPauseSafePointName(taskName string) string {
	return fmt.Sprintf("%s_pause_safepoint", taskName)
}

type PiTRTaskInfo struct {
	CheckpointInfo  *checkpoint.TaskInfoForLogRestore
	RestoreTS       uint64
	NeedFullRestore bool
}

func (p *PiTRTaskInfo) hasTiFlashItemsInCheckpoint() bool {
	return p.CheckpointInfo != nil && p.CheckpointInfo.Metadata != nil && p.CheckpointInfo.Metadata.TiFlashItems != nil
}

func (p *PiTRTaskInfo) getRestoreStartTS(ctx context.Context, pdClient pd.Client) (uint64, error) {
	if p.CheckpointInfo != nil && p.CheckpointInfo.Metadata != nil && p.CheckpointInfo.Metadata.RestoreStartTS != 0 {
		return p.CheckpointInfo.Metadata.RestoreStartTS, nil
	}
	return restore.GetTSWithRetry(ctx, pdClient)
}

func generatePiTRTaskInfo(
	ctx context.Context,
	mgr *conn.Mgr,
	g glue.Glue,
	cfg *RestoreConfig,
	restoreID uint64,
) (*PiTRTaskInfo, error) {
	var (
		doFullRestore = len(cfg.FullBackupStorage) > 0
		curTaskInfo   *checkpoint.TaskInfoForLogRestore
		err           error
	)
	checkInfo := &PiTRTaskInfo{}

	log.Info("generating PiTR task info",
		zap.Uint64("restoreID", restoreID),
		zap.Bool("useCheckpoint", cfg.UseCheckpoint),
		zap.Bool("doFullRestore", doFullRestore))

	if cfg.UseCheckpoint {
		if len(cfg.CheckpointStorage) > 0 {
			clusterID := mgr.GetPDClient().GetClusterID(ctx)
			log.Info("initializing storage checkpoint meta manager for PiTR",

Review comment (security):
Seems vulnerable. Considering a breakpoint generated before the update, seems it will allow restore a dup table ID.

Code after:
	shiftPhysical := physical - streamShiftDuration.Milliseconds()
	if shiftPhysical < 0 {
		return 0
	}
	return oracle.ComposeTS(shiftPhysical, logical)
}

func buildPauseSafePointName(taskName string) string {
	return fmt.Sprintf("%s_pause_safepoint", taskName)
}

type PiTRTaskInfo struct {
	CheckpointInfo  *checkpoint.TaskInfoForLogRestore
	RestoreTS       uint64
	NeedFullRestore bool
}

func (p *PiTRTaskInfo) hasTiFlashItemsInCheckpoint() bool {
	return p.CheckpointInfo != nil && p.CheckpointInfo.Metadata != nil && p.CheckpointInfo.Metadata.TiFlashItems != nil
}

func (p *PiTRTaskInfo) getRestoreStartTS(ctx context.Context, pdClient pd.Client) (uint64, error) {
	if p.CheckpointInfo != nil && p.CheckpointInfo.Metadata != nil {
		return p.CheckpointInfo.Metadata.RestoreStartTS, nil
	}
	return restore.GetTSWithRetry(ctx, pdClient)
}

func generatePiTRTaskInfo(
	ctx context.Context,
	mgr *conn.Mgr,
	g glue.Glue,
	cfg *RestoreConfig,
	restoreID uint64,
) (*PiTRTaskInfo, error) {
	var (
		doFullRestore = len(cfg.FullBackupStorage) > 0
		curTaskInfo   *checkpoint.TaskInfoForLogRestore
		err           error
	)
	checkInfo := &PiTRTaskInfo{}

	log.Info("generating PiTR task info",
		zap.Uint64("restoreID", restoreID),
		zap.Bool("useCheckpoint", cfg.UseCheckpoint),
		zap.Bool("doFullRestore", doFullRestore))

	if cfg.UseCheckpoint {
		if len(cfg.CheckpointStorage) > 0 {
			clusterID := mgr.GetPDClient().GetClusterID(ctx)
			log.Info("initializing storage checkpoint meta manager for PiTR",