snapper: fix delayed snapshots caused by system suspend/resume

See explainer comment in periodic.go for details.

fixes https://github.com/zrepl/zrepl/issues/611

config/config.go

@@ -6,6 +6,8 @@ import (
 	"log/syslog"
 	"os"
 	"reflect"
+	"regexp"
+	"strconv"
 	"time"
 
 	"github.com/pkg/errors"
@@ -188,10 +190,13 @@ func (i *PositiveDurationOrManual) UnmarshalYAML(u func(interface{}, bool) error
 		return fmt.Errorf("value must not be empty")
 	default:
 		i.Manual = false
-		i.Interval, err = parsePositiveDuration(s)
+		i.Interval, err = time.ParseDuration(s)
 		if err != nil {
 			return err
 		}
+		if i.Interval <= 0 {
+			return fmt.Errorf("value must be a positive duration, got %q", s)
+		}
 	}
 	return nil
 }
@@ -223,10 +228,10 @@ type SnapshottingEnum struct {
 }
 
 type SnapshottingPeriodic struct {
-	Type     string            `yaml:"type"`
-	Prefix   string            `yaml:"prefix"`
-	Interval *PositiveDuration `yaml:"interval"`
-	Hooks    HookList          `yaml:"hooks,optional"`
+	Type     string        `yaml:"type"`
+	Prefix   string        `yaml:"prefix"`
+	Interval time.Duration `yaml:"interval,positive"`
+	Hooks    HookList      `yaml:"hooks,optional"`
 }
 
 type CronSpec struct {
@@ -710,3 +715,41 @@ func ParseConfigBytes(bytes []byte) (*Config, error) {
 	}
 	return c, nil
 }
+
+var durationStringRegex *regexp.Regexp = regexp.MustCompile(`^\s*(\d+)\s*(s|m|h|d|w)\s*$`)
+
+func parsePositiveDuration(e string) (d time.Duration, err error) {
+	comps := durationStringRegex.FindStringSubmatch(e)
+	if len(comps) != 3 {
+		err = fmt.Errorf("does not match regex: %s %#v", e, comps)
+		return
+	}
+
+	durationFactor, err := strconv.ParseInt(comps[1], 10, 64)
+	if err != nil {
+		return 0, err
+	}
+	if durationFactor <= 0 {
+		return 0, errors.New("duration must be positive integer")
+	}
+
+	var durationUnit time.Duration
+	switch comps[2] {
+	case "s":
+		durationUnit = time.Second
+	case "m":
+		durationUnit = time.Minute
+	case "h":
+		durationUnit = time.Hour
+	case "d":
+		durationUnit = 24 * time.Hour
+	case "w":
+		durationUnit = 24 * 7 * time.Hour
+	default:
+		err = fmt.Errorf("contains unknown time unit '%s'", comps[2])
+		return
+	}
+
+	d = time.Duration(durationFactor) * durationUnit
+	return
+}

config/config_duration.go

@@ -1,106 +0,0 @@
-package config
-
-import (
-	"errors"
-	"fmt"
-	"regexp"
-	"strconv"
-	"time"
-
-	"github.com/kr/pretty"
-	"github.com/zrepl/yaml-config"
-)
-
-type Duration struct{ d time.Duration }
-
-func (d Duration) Duration() time.Duration { return d.d }
-
-var _ yaml.Unmarshaler = &Duration{}
-
-func (d *Duration) UnmarshalYAML(unmarshal func(v interface{}, not_strict bool) error) error {
-	var s string
-	err := unmarshal(&s, false)
-	if err != nil {
-		return err
-	}
-	d.d, err = parseDuration(s)
-	if err != nil {
-		d.d = 0
-		return &yaml.TypeError{Errors: []string{fmt.Sprintf("cannot parse value %q: %s", s, err)}}
-	}
-	return nil
-}
-
-type PositiveDuration struct{ d Duration }
-
-var _ yaml.Unmarshaler = &PositiveDuration{}
-
-func (d PositiveDuration) Duration() time.Duration { return d.d.Duration() }
-
-func (d *PositiveDuration) UnmarshalYAML(unmarshal func(v interface{}, not_strict bool) error) error {
-	err := d.d.UnmarshalYAML(unmarshal)
-	if err != nil {
-		return err
-	}
-	if d.d.Duration() <= 0 {
-		return fmt.Errorf("duration must be positive, got %s", d.d.Duration())
-	}
-	return nil
-}
-
-func parsePositiveDuration(e string) (time.Duration, error) {
-	d, err := parseDuration(e)
-	if err != nil {
-		return d, err
-	}
-	if d <= 0 {
-		return 0, errors.New("duration must be positive integer")
-	}
-	return d, err
-}
-
-var durationStringRegex *regexp.Regexp = regexp.MustCompile(`^\s*([\+-]?\d+)\s*(|s|m|h|d|w)\s*$`)
-
-func parseDuration(e string) (d time.Duration, err error) {
-	comps := durationStringRegex.FindStringSubmatch(e)
-	if comps == nil {
-		err = fmt.Errorf("must match %s", durationStringRegex)
-		return
-	}
-	if len(comps) != 3 {
-		panic(pretty.Sprint(comps))
-	}
-
-	durationFactor, err := strconv.ParseInt(comps[1], 10, 64)
-	if err != nil {
-		return 0, err
-	}
-
-	var durationUnit time.Duration
-	switch comps[2] {
-	case "":
-		if durationFactor != 0 {
-			err = fmt.Errorf("missing time unit")
-			return
-		} else {
-			// It's the case where user specified '0'.
-			// We want to allow this, just like time.ParseDuration.
-		}
-	case "s":
-		durationUnit = time.Second
-	case "m":
-		durationUnit = time.Minute
-	case "h":
-		durationUnit = time.Hour
-	case "d":
-		durationUnit = 24 * time.Hour
-	case "w":
-		durationUnit = 24 * 7 * time.Hour
-	default:
-		err = fmt.Errorf("contains unknown time unit '%s'", comps[2])
-		return
-	}
-
-	d = time.Duration(durationFactor) * durationUnit
-	return
-}

config/config_snapshotting_test.go

@@ -38,13 +38,6 @@ jobs:
     interval: 10m
 `
 
-	periodicDaily := `
-  snapshotting:
-    type: periodic
-    prefix: zrepl_
-    interval: 1d
-`
-
 	hooks := `
   snapshotting:
     type: periodic
@@ -81,15 +74,7 @@ jobs:
 		c = testValidConfig(t, fillSnapshotting(periodic))
 		snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingPeriodic)
 		assert.Equal(t, "periodic", snp.Type)
-		assert.Equal(t, 10*time.Minute, snp.Interval.Duration())
-		assert.Equal(t, "zrepl_", snp.Prefix)
-	})
-
-	t.Run("periodicDaily", func(t *testing.T) {
-		c = testValidConfig(t, fillSnapshotting(periodicDaily))
-		snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingPeriodic)
-		assert.Equal(t, "periodic", snp.Type)
-		assert.Equal(t, 24*time.Hour, snp.Interval.Duration())
+		assert.Equal(t, 10*time.Minute, snp.Interval)
 		assert.Equal(t, "zrepl_", snp.Prefix)
 	})
 

daemon/snapper/cron.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/hooks"
+	"github.com/zrepl/zrepl/util/suspendresumesafetimer"
 	"github.com/zrepl/zrepl/zfs"
 )
 
@@ -42,67 +43,45 @@ type Cron struct {
 
 func (s *Cron) Run(ctx context.Context, snapshotsTaken chan<- struct{}) {
 
-	t := time.NewTimer(0)
-	defer func() {
-		if !t.Stop() {
-			select {
-			case <-t.C:
-			default:
-			}
-		}
-	}()
 	for {
 		now := time.Now()
 		s.mtx.Lock()
 		s.wakeupTime = s.config.Cron.Schedule.Next(now)
 		s.mtx.Unlock()
 
-		// Re-arm the timer.
-		// Need to Stop before Reset, see docs.
-		if !t.Stop() {
-			// Use non-blocking read from timer channel
-			// because, except for the first loop iteration,
-			// the channel is already drained
-			select {
-			case <-t.C:
-			default:
-			}
-		}
-		t.Reset(s.wakeupTime.Sub(now))
-
-		select {
-		case <-ctx.Done():
+		ctxDone := suspendresumesafetimer.SleepUntil(ctx, s.wakeupTime)
+		if ctxDone != nil {
 			return
-		case <-t.C:
-			getLogger(ctx).Debug("cron timer fired")
-			s.mtx.Lock()
-			if s.running {
-				getLogger(ctx).Warn("snapshotting triggered according to cron rules but previous snapshotting is not done; not taking a snapshot this time")
-				s.wakeupWhileRunningCount++
-				s.mtx.Unlock()
-				continue
-			}
-			s.lastError = nil
-			s.lastPlan = nil
-			s.wakeupWhileRunningCount = 0
-			s.running = true
-			s.mtx.Unlock()
-			go func() {
-				err := s.do(ctx)
-				s.mtx.Lock()
-				s.lastError = err
-				s.running = false
-				s.mtx.Unlock()
-
-				select {
-				case snapshotsTaken <- struct{}{}:
-				default:
-					if snapshotsTaken != nil {
-						getLogger(ctx).Warn("callback channel is full, discarding snapshot update event")
-					}
-				}
-			}()
 		}
+
+		getLogger(ctx).Debug("cron timer fired")
+		s.mtx.Lock()
+		if s.running {
+			getLogger(ctx).Warn("snapshotting triggered according to cron rules but previous snapshotting is not done; not taking a snapshot this time")
+			s.wakeupWhileRunningCount++
+			s.mtx.Unlock()
+			continue
+		}
+		s.lastError = nil
+		s.lastPlan = nil
+		s.wakeupWhileRunningCount = 0
+		s.running = true
+		s.mtx.Unlock()
+		go func() {
+			err := s.do(ctx)
+			s.mtx.Lock()
+			s.lastError = err
+			s.running = false
+			s.mtx.Unlock()
+
+			select {
+			case snapshotsTaken <- struct{}{}:
+			default:
+				if snapshotsTaken != nil {
+					getLogger(ctx).Warn("callback channel is full, discarding snapshot update event")
+				}
+			}
+		}()
 	}
 
 }

daemon/snapper/periodic.go

@@ -15,6 +15,7 @@ import (
 	"github.com/zrepl/zrepl/daemon/hooks"
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/util/envconst"
+	"github.com/zrepl/zrepl/util/suspendresumesafetimer"
 	"github.com/zrepl/zrepl/zfs"
 )
 
@@ -22,7 +23,7 @@ func periodicFromConfig(g *config.Global, fsf zfs.DatasetFilter, in *config.Snap
 	if in.Prefix == "" {
 		return nil, errors.New("prefix must not be empty")
 	}
-	if in.Interval.Duration() <= 0 {
+	if in.Interval <= 0 {
 		return nil, errors.New("interval must be positive")
 	}
 
@@ -32,7 +33,7 @@ func periodicFromConfig(g *config.Global, fsf zfs.DatasetFilter, in *config.Snap
 	}
 
 	args := periodicArgs{
-		interval: in.Interval.Duration(),
+		interval: in.Interval,
 		fsf:      fsf,
 		planArgs: planArgs{
 			prefix: in.Prefix,
@@ -171,16 +172,13 @@ func periodicStateSyncUp(a periodicArgs, u updater) state {
 	u(func(s *Periodic) {
 		s.sleepUntil = syncPoint
 	})
-	t := time.NewTimer(time.Until(syncPoint))
-	defer t.Stop()
-	select {
-	case <-t.C:
-		return u(func(s *Periodic) {
-			s.state = Planning
-		}).sf()
-	case <-a.ctx.Done():
+	ctxDone := suspendresumesafetimer.SleepUntil(a.ctx, syncPoint)
+	if ctxDone != nil {
 		return onMainCtxDone(a.ctx, u)
 	}
+	return u(func(s *Periodic) {
+		s.state = Planning
+	}).sf()
 }
 
 func periodicStatePlan(a periodicArgs, u updater) state {
@@ -241,17 +239,13 @@ func periodicStateWait(a periodicArgs, u updater) state {
 		logFunc("enter wait-state after error")
 	})
 
-	t := time.NewTimer(time.Until(sleepUntil))
-	defer t.Stop()
-
-	select {
-	case <-t.C:
-		return u(func(snapper *Periodic) {
-			snapper.state = Planning
-		}).sf()
-	case <-a.ctx.Done():
+	ctxDone := suspendresumesafetimer.SleepUntil(a.ctx, sleepUntil)
+	if ctxDone != nil {
 		return onMainCtxDone(a.ctx, u)
 	}
+	return u(func(snapper *Periodic) {
+		snapper.state = Planning
+	}).sf()
 }
 
 func listFSes(ctx context.Context, mf zfs.DatasetFilter) (fss []*zfs.DatasetPath, err error) {

util/suspendresumesafetimer/suspendresumesafetimer.go

@@ -0,0 +1,96 @@
+package suspendresumesafetimer
+
+import (
+	"context"
+	"time"
+
+	"github.com/zrepl/zrepl/util/envconst"
+)
+
+// The returned error is guaranteed to be the ctx.Err()
+func SleepUntil(ctx context.Context, sleepUntil time.Time) error {
+
+	// We use .Round(0) to strip the monotonic clock reading from the time.Time
+	// returned by time.Now(). That will make the before/after check in the ticker
+	// for-loop compare wall-clock times instead of monotonic time.
+	// Comparing wall clock time is necessary because monotonic time does not progress
+	// while the system is suspended.
+	//
+	// Background
+	//
+	// A time.Time carries a wallclock timestamp and optionally a monotonic clock timestamp.
+	// time.Now() returns a time.Time that carries both.
+	// time.Time.Add() applies the same delta to both timestamps in the time.Time.
+	// x.Sub(y) will return the *monotonic* delta if both x and y carry a monotonic timestamp.
+	// time.Until(x) == x.Sub(now) where `now` will have a monotonic timestamp.
+	// So, time.Until(x) with an `x` that has monotonic timestamp will return monotonic delta.
+	//
+	// Why Do We Care?
+	//
+	// On systems that suspend/resume, wall clock time progresses during suspend but
+	// monotonic time does not.
+	//
+	// So, suppose the following sequence of events:
+	//   x <== time.Now()
+	//   System suspends for 1 hour
+	//   delta <== time.Now().Sub(x)
+	// `delta` will be near 0 because time.Until() subtracts the monotonic
+	// timestamps, and monotonic time didn't progress during suspend.
+	//
+	// Now strip the timestamp using .Round(0)
+	//   x <== time.Now().Round(0)
+	//   System suspends for 1 hour
+	//   delta <== time.Now().Sub(x)
+	// `delta` will be 1 hour because time.Sub() subtracted wallclock timestamps
+	// because x didn't have a monotonic timestamp because we stripped it using .Round(0).
+	//
+	//
+	sleepUntil = sleepUntil.Round(0)
+
+	// Set up a timer so that, if the system doesn't suspend/resume,
+	// we get a precise wake-up time from the native Go timer.
+	monotonicClockTimer := time.NewTimer(time.Until(sleepUntil))
+	defer func() {
+		if !monotonicClockTimer.Stop() {
+			// non-blocking read since we can come here when
+			// we've already drained the channel through
+			//		case <-monotonicClockTimer.C
+			// in the `for` loop below.
+			select {
+			case <-monotonicClockTimer.C:
+			default:
+			}
+		}
+	}()
+
+	// Set up a ticker so that we're guaranteed to wake up periodically.
+	// We'll then get the current wall-clock time and check ourselves
+	// whether we're past the requested expiration time.
+	// Pick a 10 second check interval by default since it's rare that
+	// suspend/resume is done more frequently.
+	ticker := time.NewTicker(envconst.Duration("ZREPL_WALLCLOCKTIMER_MAX_DELAY", 10*time.Second))
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-monotonicClockTimer.C:
+			return nil
+		case <-ticker.C:
+			now := time.Now()
+			if now.Before(sleepUntil) {
+				// Continue waiting.
+
+				// Reset the monotonic timer to reset drift.
+				if !monotonicClockTimer.Stop() {
+					<-monotonicClockTimer.C
+				}
+				monotonicClockTimer.Reset(time.Until(sleepUntil))
+
+				continue
+			}
+			return nil
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}