diff --git a/replication/fsrep/fsfsm.go b/replication/fsrep/fsfsm.go index facb889..3c84b15 100644 --- a/replication/fsrep/fsfsm.go +++ b/replication/fsrep/fsfsm.go @@ -82,6 +82,7 @@ const ( type Error interface { error Temporary() bool + ContextErr() bool LocalToFS() bool } @@ -187,6 +188,8 @@ func (e *ReplicationConflictError) Error() string { return fmt.Sprintf("permanen func (e *ReplicationConflictError) LocalToFS() bool { return true } +func (e *ReplicationConflictError) ContextErr() bool { return false } + func NewReplicationConflictError(fs string, err error) *Replication { return &Replication{ state: Completed, @@ -332,6 +335,16 @@ func (e StepError) LocalToFS() bool { return true // conservative approximation: we'd like to check for specific errors returned over RPC here... } +func (e StepError) ContextErr() bool { + switch e.err { + case context.Canceled: + return true + case context.DeadlineExceeded: + return true + } + return false +} + func (fsr *Replication) Report() *Report { fsr.lock.Lock() defer fsr.lock.Unlock() diff --git a/replication/mainfsm.go b/replication/mainfsm.go index 7a5ba12..2c8de45 100644 --- a/replication/mainfsm.go +++ b/replication/mainfsm.go @@ -459,7 +459,14 @@ func stateWorking(ctx context.Context, ka *watchdog.KeepAlive, sender Sender, re }).rsf() if err != nil { - if err.LocalToFS() { + if err.ContextErr() && ctx.Err() != nil { + getLogger(ctx).WithError(err). + Info("filesystem replication was cancelled") + u(func(r*Replication) { + r.err = GlobalError{Err: err, Temporary: false} + r.state = PermanentError + }) + } else if err.LocalToFS() { getLogger(ctx).WithError(err). Error("filesystem replication encountered a filesystem-specific error") // we stay in this state and let the queuing logic above de-prioritize this failing FS