feat(state): crossport latest changes from snapd/overlord/state (#344)

There are several major changes to the state package:

* Add WaitStatus support that allows a task to wait until further action
  to continue its execution. The WaitStatus is treated mostly as
  DoneStatus, except it is not ready status.
* Add Change.AbortUnreadyLanes.
* Add Change.CheckTaskDependencies to check if tasks have circular
  dependencies.
* Add task and change callbacks invoked on a status change.
* Update clients of the State.Get to use a new NoStateError to check if
  a desired key is present.
* Take StartOfOperationTime as a Prune parameter.

internals/daemon/daemon.go

@@ -690,7 +690,7 @@ func (d *Daemon) rebootDelay() (time.Duration, error) {
 	// see whether a reboot had already been scheduled
 	var rebootAt time.Time
 	err := d.state.Get("daemon-system-restart-at", &rebootAt)
-	if err != nil && err != state.ErrNoState {
+	if err != nil && !errors.Is(err, state.ErrNoState) {
 		return 0, err
 	}
 	rebootDelay := 1 * time.Minute
@@ -832,7 +832,7 @@ var errExpectedReboot = errors.New("expected reboot did not happen")
 func (d *Daemon) RebootIsMissing(st *state.State) error {
 	var nTentative int
 	err := st.Get("daemon-system-restart-tentative", &nTentative)
-	if err != nil && err != state.ErrNoState {
+	if err != nil && !errors.Is(err, state.ErrNoState) {
 		return err
 	}
 	nTentative++

internals/daemon/daemon_test.go

@@ -991,8 +991,8 @@ func (s *daemonSuite) TestRestartExpectedRebootOK(c *C) {
 	defer st.Unlock()
 	var v interface{}
 	// these were cleared
-	c.Check(st.Get("daemon-system-restart-at", &v), Equals, state.ErrNoState)
-	c.Check(st.Get("system-restart-from-boot-id", &v), Equals, state.ErrNoState)
+	c.Check(st.Get("daemon-system-restart-at", &v), testutil.ErrorIs, state.ErrNoState)
+	c.Check(st.Get("system-restart-from-boot-id", &v), testutil.ErrorIs, state.ErrNoState)
 }
 
 func (s *daemonSuite) TestRestartExpectedRebootGiveUp(c *C) {
@@ -1015,9 +1015,9 @@ func (s *daemonSuite) TestRestartExpectedRebootGiveUp(c *C) {
 	defer st.Unlock()
 	var v interface{}
 	// these were cleared
-	c.Check(st.Get("daemon-system-restart-at", &v), Equals, state.ErrNoState)
-	c.Check(st.Get("system-restart-from-boot-id", &v), Equals, state.ErrNoState)
-	c.Check(st.Get("daemon-system-restart-tentative", &v), Equals, state.ErrNoState)
+	c.Check(st.Get("daemon-system-restart-at", &v), testutil.ErrorIs, state.ErrNoState)
+	c.Check(st.Get("system-restart-from-boot-id", &v), testutil.ErrorIs, state.ErrNoState)
+	c.Check(st.Get("daemon-system-restart-tentative", &v), testutil.ErrorIs, state.ErrNoState)
 }
 
 func (s *daemonSuite) TestRestartIntoSocketModeNoNewChanges(c *C) {

internals/overlord/export_test.go

@@ -40,6 +40,14 @@ func FakePruneInterval(prunei, prunew, abortw time.Duration) (restore func()) {
 	}
 }
 
+func FakePruneTicker(f func(t *time.Ticker) <-chan time.Time) (restore func()) {
+	old := pruneTickerC
+	pruneTickerC = f
+	return func() {
+		pruneTickerC = old
+	}
+}
+
 // FakeEnsureNext sets o.ensureNext for tests.
 func FakeEnsureNext(o *Overlord, t time.Time) {
 	o.ensureNext = t
@@ -49,3 +57,11 @@ func FakeEnsureNext(o *Overlord, t time.Time) {
 func (o *Overlord) Engine() *StateEngine {
 	return o.stateEng
 }
+
+func FakeTimeNow(f func() time.Time) (restore func()) {
+	old := timeNow
+	timeNow = f
+	return func() {
+		timeNow = old
+	}
+}

internals/overlord/overlord.go

@@ -16,6 +16,7 @@
 package overlord
 
 import (
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -49,6 +50,10 @@ var (
 	defaultCachedDownloads = 5
 )
 
+var pruneTickerC = func(t *time.Ticker) <-chan time.Time {
+	return t.C
+}
+
 // Extension represents an extension of the Overlord.
 type Extension interface {
 	// ExtraManagers allows additional StateManagers to be used.
@@ -81,6 +86,8 @@ type Overlord struct {
 	ensureRun   int32
 	pruneTicker *time.Ticker
 
+	startOfOperationTime time.Time
+
 	// managers
 	inited     bool
 	startedUp  bool
@@ -258,6 +265,15 @@ func (o *Overlord) StartUp() error {
 		return nil
 	}
 	o.startedUp = true
+
+	var err error
+	st := o.State()
+	st.Lock()
+	o.startOfOperationTime, err = o.StartOfOperationTime()
+	st.Unlock()
+	if err != nil {
+		return fmt.Errorf("cannot get start of operation time: %s", err)
+	}
 	return o.stateEng.StartUp()
 }
 
@@ -314,14 +330,15 @@ func (o *Overlord) Loop() {
 			// continue to the next Ensure() try for now
 			o.stateEng.Ensure()
 			o.ensureDidRun()
+			pruneC := pruneTickerC(o.pruneTicker)
 			select {
 			case <-o.loopTomb.Dying():
 				return nil
 			case <-o.ensureTimer.C:
-			case <-o.pruneTicker.C:
+			case <-pruneC:
 				st := o.State()
 				st.Lock()
-				st.Prune(pruneWait, abortWait, pruneMaxChanges)
+				st.Prune(o.startOfOperationTime, pruneWait, abortWait, pruneMaxChanges)
 				st.Unlock()
 			}
 		}
@@ -499,6 +516,23 @@ func (o *Overlord) AddManager(mgr StateManager) {
 	o.stateEng.AddManager(mgr)
 }
 
+var timeNow = time.Now
+
+func (m *Overlord) StartOfOperationTime() (time.Time, error) {
+	var opTime time.Time
+	err := m.State().Get("start-of-operation-time", &opTime)
+	if err == nil {
+		return opTime, nil
+	}
+	if err != nil && !errors.Is(err, state.ErrNoState) {
+		return opTime, err
+	}
+	opTime = timeNow()
+
+	m.State().Set("start-of-operation-time", opTime)
+	return opTime, nil
+}
+
 type fakeBackend struct {
 	o *Overlord
 }

internals/overlord/overlord_test.go

@@ -46,6 +46,26 @@ type overlordSuite struct {
 
 var _ = Suite(&overlordSuite{})
 
+type ticker struct {
+	tickerChannel chan time.Time
+}
+
+func (w *ticker) tick(n int) {
+	for i := 0; i < n; i++ {
+		w.tickerChannel <- time.Now()
+	}
+}
+
+func fakePruneTicker() (w *ticker, restore func()) {
+	w = &ticker{
+		tickerChannel: make(chan time.Time),
+	}
+	restore = overlord.FakePruneTicker(func(t *time.Ticker) <-chan time.Time {
+		return w.tickerChannel
+	})
+	return w, restore
+}
+
 func (ovs *overlordSuite) SetUpTest(c *C) {
 	ovs.dir = c.MkDir()
 	ovs.statePath = filepath.Join(ovs.dir, ".pebble.state")
@@ -517,6 +537,137 @@ func (ovs *overlordSuite) TestEnsureLoopPruneRunsMultipleTimes(c *C) {
 	c.Assert(err, IsNil)
 }
 
+func (ovs *overlordSuite) TestOverlordStartUpSetsStartOfOperation(c *C) {
+	restoreIntv := overlord.FakePruneInterval(100*time.Millisecond, 1000*time.Millisecond, 1*time.Hour)
+	defer restoreIntv()
+
+	// use real overlord, we need device manager to be there
+	o, err := overlord.New(&overlord.Options{PebbleDir: ovs.dir})
+	c.Assert(err, IsNil)
+
+	st := o.State()
+	st.Lock()
+	defer st.Unlock()
+
+	// validity check, not set
+	var opTime time.Time
+	c.Assert(st.Get("start-of-operation-time", &opTime), testutil.ErrorIs, state.ErrNoState)
+	st.Unlock()
+
+	c.Assert(o.StartUp(), IsNil)
+
+	st.Lock()
+	c.Assert(st.Get("start-of-operation-time", &opTime), IsNil)
+}
+
+func (ovs *overlordSuite) TestEnsureLoopPruneDoesntAbortShortlyAfterStartOfOperation(c *C) {
+	w, restoreTicker := fakePruneTicker()
+	defer restoreTicker()
+
+	// use real overlord, we need device manager to be there
+	o, err := overlord.New(&overlord.Options{PebbleDir: ovs.dir})
+	c.Assert(err, IsNil)
+
+	// avoid immediate transition to Done due to unknown kind
+	o.TaskRunner().AddHandler("bar", func(t *state.Task, _ *tomb.Tomb) error {
+		return &state.Retry{}
+	}, nil)
+
+	st := o.State()
+	st.Lock()
+
+	// start of operation time is 50min ago, this is less then abort limit
+	opTime := time.Now().Add(-50 * time.Minute)
+	st.Set("start-of-operation-time", opTime)
+
+	// spawn time one month ago
+	spawnTime := time.Now().AddDate(0, -1, 0)
+	restoreTimeNow := state.FakeTime(spawnTime)
+
+	t := st.NewTask("bar", "...")
+	chg := st.NewChange("other-change", "...")
+	chg.AddTask(t)
+
+	restoreTimeNow()
+
+	// validity
+	c.Check(st.Changes(), HasLen, 1)
+
+	st.Unlock()
+	c.Assert(o.StartUp(), IsNil)
+
+	// start the loop that runs the prune ticker
+	o.Loop()
+	w.tick(2)
+
+	c.Assert(o.Stop(), IsNil)
+
+	st.Lock()
+	defer st.Unlock()
+	c.Assert(st.Changes(), HasLen, 1)
+	c.Check(chg.Status(), Equals, state.DoingStatus)
+}
+
+func (ovs *overlordSuite) TestEnsureLoopPruneAbortsOld(c *C) {
+	// Ensure interval is not relevant for this test
+	restoreEnsureIntv := overlord.FakeEnsureInterval(10 * time.Hour)
+	defer restoreEnsureIntv()
+
+	w, restoreTicker := fakePruneTicker()
+	defer restoreTicker()
+
+	// use real overlord, we need device manager to be there
+	o, err := overlord.New(&overlord.Options{PebbleDir: ovs.dir})
+	c.Assert(err, IsNil)
+
+	// avoid immediate transition to Done due to having unknown kind
+	o.TaskRunner().AddHandler("bar", func(t *state.Task, _ *tomb.Tomb) error {
+		return &state.Retry{}
+	}, nil)
+
+	st := o.State()
+	st.Lock()
+
+	// start of operation time is a year ago
+	opTime := time.Now().AddDate(-1, 0, 0)
+	st.Set("start-of-operation-time", opTime)
+
+	st.Unlock()
+	c.Assert(o.StartUp(), IsNil)
+	st.Lock()
+
+	// spawn time one month ago
+	spawnTime := time.Now().AddDate(0, -1, 0)
+	restoreTimeNow := state.FakeTime(spawnTime)
+	t := st.NewTask("bar", "...")
+	chg := st.NewChange("other-change", "...")
+	chg.AddTask(t)
+
+	restoreTimeNow()
+
+	// validity
+	c.Check(st.Changes(), HasLen, 1)
+	st.Unlock()
+
+	// start the loop that runs the prune ticker
+	o.Loop()
+	w.tick(2)
+
+	c.Assert(o.Stop(), IsNil)
+
+	st.Lock()
+	defer st.Unlock()
+
+	// validity
+	op, err := o.StartOfOperationTime()
+	c.Assert(err, IsNil)
+	c.Check(op.Equal(opTime), Equals, true)
+
+	c.Assert(st.Changes(), HasLen, 1)
+	// change was aborted
+	c.Check(chg.Status(), Equals, state.HoldStatus)
+}
+
 func (ovs *overlordSuite) TestCheckpoint(c *C) {
 	oldUmask := syscall.Umask(0)
 	defer syscall.Umask(oldUmask)
@@ -907,3 +1058,40 @@ func (ovs *overlordSuite) TestOverlordCanStandby(c *C) {
 
 	c.Assert(o.CanStandby(), Equals, true)
 }
+
+func (ovs *overlordSuite) TestStartOfOperationTimeAlreadySet(c *C) {
+	o := overlord.Fake()
+	st := o.State()
+	st.Lock()
+	defer st.Unlock()
+
+	op := time.Now().AddDate(0, -1, 0)
+	st.Set("start-of-operation-time", op)
+
+	operationTime, err := o.StartOfOperationTime()
+	c.Assert(err, IsNil)
+	c.Check(operationTime.Equal(op), Equals, true)
+}
+
+func (s *overlordSuite) TestStartOfOperationSetTime(c *C) {
+	o := overlord.Fake()
+	st := o.State()
+	st.Lock()
+	defer st.Unlock()
+
+	now := time.Now().Add(-1 * time.Second)
+	overlord.FakeTimeNow(func() time.Time {
+		return now
+	})
+
+	operationTime, err := o.StartOfOperationTime()
+	c.Assert(err, IsNil)
+	c.Check(operationTime.Equal(now), Equals, true)
+
+	// repeated call returns already set time
+	prev := now
+	now = time.Now().Add(-10 * time.Hour)
+	operationTime, err = o.StartOfOperationTime()
+	c.Assert(err, IsNil)
+	c.Check(operationTime.Equal(prev), Equals, true)
+}

internals/overlord/patch/patch.go

@@ -20,6 +20,7 @@
 package patch
 
 import (
+	"errors"
 	"fmt"
 
 	"github.com/canonical/pebble/internals/logger"
@@ -43,12 +44,12 @@ var patches = make(map[int][]PatchFunc)
 func Init(s *state.State) {
 	s.Lock()
 	defer s.Unlock()
-	if s.Get("patch-level", new(int)) != state.ErrNoState {
+	if err := s.Get("patch-level", new(int)); !errors.Is(err, state.ErrNoState) {
 		panic("internal error: expected empty state, attempting to override patch-level without actual patching")
 	}
 	s.Set("patch-level", Level)
 
-	if s.Get("patch-sublevel", new(int)) != state.ErrNoState {
+	if err := s.Get("patch-sublevel", new(int)); !errors.Is(err, state.ErrNoState) {
 		panic("internal error: expected empty state, attempting to override patch-sublevel without actual patching")
 	}
 	s.Set("patch-sublevel", Sublevel)
@@ -76,12 +77,12 @@ func Apply(s *state.State) error {
 	var stateLevel, stateSublevel int
 	s.Lock()
 	err := s.Get("patch-level", &stateLevel)
-	if err == nil || err == state.ErrNoState {
+	if err == nil || errors.Is(err, state.ErrNoState) {
 		err = s.Get("patch-sublevel", &stateSublevel)
 	}
 	s.Unlock()
 
-	if err != nil && err != state.ErrNoState {
+	if err != nil && !errors.Is(err, state.ErrNoState) {
 		return err
 	}