mirror of
https://gitee.com/dromara/mayfly-go
synced 2025-11-02 15:30:25 +08:00
!85 fix: 修复 BINLOG同步任务加载问题
* Merge branch 'dev' of gitee.com:dromara/mayfly-go into feat-db-bak * fix: 修复 BINLOG 同步任务加载问题
This commit is contained in:
@@ -24,9 +24,11 @@ func (b *DbBackup) GetDbName() string {
|
||||
}
|
||||
|
||||
func (b *DbBackup) Schedule() (time.Time, error) {
|
||||
var deadline time.Time
|
||||
if b.IsFinished() || !b.Enabled {
|
||||
return deadline, runner.ErrFinished
|
||||
if b.IsFinished() {
|
||||
return time.Time{}, runner.ErrJobFinished
|
||||
}
|
||||
if !b.Enabled {
|
||||
return time.Time{}, runner.ErrJobDisabled
|
||||
}
|
||||
switch b.LastStatus {
|
||||
case DbJobSuccess:
|
||||
@@ -34,13 +36,12 @@ func (b *DbBackup) Schedule() (time.Time, error) {
|
||||
if lastTime.Before(b.StartTime) {
|
||||
lastTime = b.StartTime.Add(-b.Interval)
|
||||
}
|
||||
deadline = lastTime.Add(b.Interval - lastTime.Sub(b.StartTime)%b.Interval)
|
||||
return lastTime.Add(b.Interval - lastTime.Sub(b.StartTime)%b.Interval), nil
|
||||
case DbJobFailed:
|
||||
deadline = time.Now().Add(time.Minute)
|
||||
return time.Now().Add(time.Minute), nil
|
||||
default:
|
||||
deadline = b.StartTime
|
||||
return b.StartTime, nil
|
||||
}
|
||||
return deadline, nil
|
||||
}
|
||||
|
||||
func (b *DbBackup) IsFinished() bool {
|
||||
|
||||
@@ -43,9 +43,8 @@ func (b *DbBinlog) GetDbName() string {
|
||||
func (b *DbBinlog) Schedule() (time.Time, error) {
|
||||
switch b.GetJobBase().LastStatus {
|
||||
case DbJobSuccess:
|
||||
return time.Time{}, runner.ErrFinished
|
||||
return time.Time{}, runner.ErrJobFinished
|
||||
case DbJobFailed:
|
||||
|
||||
return time.Now().Add(BinlogDownloadInterval), nil
|
||||
default:
|
||||
return time.Now(), nil
|
||||
|
||||
@@ -28,23 +28,18 @@ func (r *DbRestore) GetDbName() string {
|
||||
}
|
||||
|
||||
func (r *DbRestore) Schedule() (time.Time, error) {
|
||||
var deadline time.Time
|
||||
if r.IsFinished() || !r.Enabled {
|
||||
return deadline, runner.ErrFinished
|
||||
if !r.Enabled {
|
||||
return time.Time{}, runner.ErrJobDisabled
|
||||
}
|
||||
switch r.LastStatus {
|
||||
case DbJobSuccess:
|
||||
lastTime := r.LastTime.Time
|
||||
if lastTime.Before(r.StartTime) {
|
||||
lastTime = r.StartTime.Add(-r.Interval)
|
||||
}
|
||||
deadline = lastTime.Add(r.Interval - lastTime.Sub(r.StartTime)%r.Interval)
|
||||
case DbJobFailed:
|
||||
deadline = time.Now().Add(time.Minute)
|
||||
case DbJobSuccess, DbJobFailed:
|
||||
return time.Time{}, runner.ErrJobFinished
|
||||
default:
|
||||
deadline = r.StartTime
|
||||
if time.Now().Sub(r.StartTime) > time.Hour {
|
||||
return time.Time{}, runner.ErrJobTimeout
|
||||
}
|
||||
return r.StartTime, nil
|
||||
}
|
||||
return deadline, nil
|
||||
}
|
||||
|
||||
func (r *DbRestore) IsEnabled() bool {
|
||||
|
||||
@@ -22,11 +22,13 @@ func NewDbBackupRepo() repository.DbBackup {
|
||||
|
||||
func (d *dbBackupRepoImpl) GetDbNamesWithoutBackup(instanceId uint64, dbNames []string) ([]string, error) {
|
||||
var dbNamesWithBackup []string
|
||||
query := gormx.NewQuery(d.GetModel()).
|
||||
Eq("db_instance_id", instanceId).
|
||||
Eq("repeated", true).
|
||||
Undeleted()
|
||||
if err := query.GenGdb().Pluck("db_name", &dbNamesWithBackup).Error; err != nil {
|
||||
err := global.Db.Model(d.GetModel()).
|
||||
Where("db_instance_id = ?", instanceId).
|
||||
Where("repeated = ?", true).
|
||||
Scopes(gormx.UndeleteScope).
|
||||
Pluck("db_name", &dbNamesWithBackup).
|
||||
Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result := make([]string, 0, len(dbNames))
|
||||
@@ -39,11 +41,13 @@ func (d *dbBackupRepoImpl) GetDbNamesWithoutBackup(instanceId uint64, dbNames []
|
||||
}
|
||||
|
||||
func (d *dbBackupRepoImpl) ListDbInstances(enabled bool, repeated bool, instanceIds *[]uint64) error {
|
||||
query := gormx.NewQuery(d.GetModel()).
|
||||
Eq0("enabled", enabled).
|
||||
Eq0("repeated", repeated).
|
||||
Undeleted()
|
||||
return query.GenGdb().Distinct().Pluck("db_instance_id", &instanceIds).Error
|
||||
return global.Db.Model(d.GetModel()).
|
||||
Where("enabled = ?", enabled).
|
||||
Where("repeated = ?", repeated).
|
||||
Scopes(gormx.UndeleteScope).
|
||||
Distinct().
|
||||
Pluck("db_instance_id", &instanceIds).
|
||||
Error
|
||||
}
|
||||
|
||||
func (d *dbBackupRepoImpl) ListToDo(jobs any) error {
|
||||
|
||||
@@ -22,13 +22,16 @@ func NewDbRestoreRepo() repository.DbRestore {
|
||||
|
||||
func (d *dbRestoreRepoImpl) GetDbNamesWithoutRestore(instanceId uint64, dbNames []string) ([]string, error) {
|
||||
var dbNamesWithRestore []string
|
||||
query := gormx.NewQuery(d.GetModel()).
|
||||
Eq("db_instance_id", instanceId).
|
||||
Eq("repeated", true).
|
||||
Undeleted()
|
||||
if err := query.GenGdb().Pluck("db_name", &dbNamesWithRestore).Error; err != nil {
|
||||
err := global.Db.Model(d.GetModel()).
|
||||
Where("db_instance_id = ?", instanceId).
|
||||
Where("repeated = ?", true).
|
||||
Scopes(gormx.UndeleteScope).
|
||||
Pluck("db_name", &dbNamesWithRestore).
|
||||
Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
result := make([]string, 0, len(dbNames))
|
||||
for _, name := range dbNames {
|
||||
if !slices.Contains(dbNamesWithRestore, name) {
|
||||
|
||||
@@ -11,9 +11,11 @@ import (
|
||||
)
|
||||
|
||||
var (
|
||||
ErrNotFound = errors.New("job not found")
|
||||
ErrExist = errors.New("job already exists")
|
||||
ErrFinished = errors.New("job already finished")
|
||||
ErrJobNotFound = errors.New("job not found")
|
||||
ErrJobExist = errors.New("job already exists")
|
||||
ErrJobFinished = errors.New("job already finished")
|
||||
ErrJobDisabled = errors.New("job has been disabled")
|
||||
ErrJobTimeout = errors.New("job has timed out")
|
||||
)
|
||||
|
||||
type JobKey = string
|
||||
@@ -236,7 +238,7 @@ func (r *Runner[T]) afterRun(wrap *wrapper[T]) {
|
||||
func (r *Runner[T]) doScheduleJob(job T, finished bool) (time.Time, error) {
|
||||
if r.scheduleJob == nil {
|
||||
if finished {
|
||||
return time.Time{}, ErrFinished
|
||||
return time.Time{}, ErrJobFinished
|
||||
}
|
||||
return time.Now(), nil
|
||||
}
|
||||
@@ -293,7 +295,7 @@ func (r *Runner[T]) Add(ctx context.Context, job T) error {
|
||||
defer r.mutex.Unlock()
|
||||
|
||||
if _, ok := r.all[job.GetKey()]; ok {
|
||||
return ErrExist
|
||||
return ErrJobExist
|
||||
}
|
||||
deadline, err := r.doScheduleJob(job, false)
|
||||
if err != nil {
|
||||
@@ -358,7 +360,7 @@ func (r *Runner[T]) Remove(ctx context.Context, key JobKey) error {
|
||||
|
||||
wrap, ok := r.all[key]
|
||||
if !ok {
|
||||
return ErrNotFound
|
||||
return ErrJobNotFound
|
||||
}
|
||||
switch wrap.status {
|
||||
case JobDelaying:
|
||||
|
||||
@@ -75,7 +75,7 @@ func TestRunner_AddJob(t *testing.T) {
|
||||
{
|
||||
name: "repetitive job",
|
||||
job: newTestJob("dual"),
|
||||
want: ErrExist,
|
||||
want: ErrJobExist,
|
||||
},
|
||||
}
|
||||
runner := NewRunner[*testJob](1, func(ctx context.Context, job *testJob) {
|
||||
|
||||
Reference in New Issue
Block a user