o/registrystate: add registry tx task handlers and helpers (canonical…

…#14511)

* o/registrystate: add registry tx task handlers and helpers

Adds a RegistryManager which implements handlers for tasks related to
the committing of registry transactions. These do things like committing
the transaction internally, clean up state once that's done or on error,
etc. This doesn't include logic to deal with concurrent writers, that
will be added later once the registrystate side of this is added as
well.

Signed-off-by: Miguel Pires <miguel.pires@canonical.com>

overlord/registrystate/export_test.go

@@ -30,6 +30,8 @@ var (
 	GetPlugsAffectedByPaths   = getPlugsAffectedByPaths
 	CreateChangeRegistryTasks = createChangeRegistryTasks
 	SetTransaction            = setTransaction
+	SetOngoingTransaction     = setOngoingTransaction
+	UnsetOngoingTransaction   = unsetOngoingTransaction
 )
 
 func ChangeViewHandlerGenerator(ctx *hookstate.Context) hookstate.Handler {

overlord/registrystate/registrymgr.go

@@ -28,14 +28,32 @@ import (
 	"github.com/snapcore/snapd/logger"
 	"github.com/snapcore/snapd/overlord/hookstate"
 	"github.com/snapcore/snapd/overlord/state"
+	"gopkg.in/tomb.v2"
 )
 
+func setupRegistryHook(st *state.State, snapName, hookName string, ignoreError bool) *state.Task {
+	hookSup := &hookstate.HookSetup{
+		Snap:        snapName,
+		Hook:        hookName,
+		Optional:    true,
+		IgnoreError: ignoreError,
+	}
+	summary := fmt.Sprintf(i18n.G("Run hook %s of snap %q"), hookName, snapName)
+	task := hookstate.HookTask(st, summary, hookSup, nil)
+	return task
+}
+
 type RegistryManager struct{}
 
-func Manager(st *state.State, hookMgr *hookstate.HookManager, _ *state.TaskRunner) *RegistryManager {
+func Manager(st *state.State, hookMgr *hookstate.HookManager, runner *state.TaskRunner) *RegistryManager {
 	m := &RegistryManager{}
 
-	// TODO: add task handlers (commit-transaction, clear-state, etc)
+	// no undo since if we commit there's no rolling back
+	runner.AddHandler("commit-registry-tx", m.doCommitTransaction, nil)
+	// only activated on undo, to clear the ongoing transaction from state and
+	// unblock others who may be waiting for it
+	runner.AddHandler("clear-registry-tx-on-error", m.noop, m.clearOngoingTransaction)
+	runner.AddHandler("clear-registry-tx", m.clearOngoingTransaction, nil)
 
 	hookMgr.Register(regexp.MustCompile("^change-view-.+$"), func(context *hookstate.Context) hookstate.Handler {
 		return &changeViewHandler{ctx: context}
@@ -52,18 +70,95 @@ func Manager(st *state.State, hookMgr *hookstate.HookManager, _ *state.TaskRunne
 
 func (m *RegistryManager) Ensure() error { return nil }
 
-func setupRegistryHook(st *state.State, snapName, hookName string, ignoreError bool) *state.Task {
-	hookSup := &hookstate.HookSetup{
-		Snap:        snapName,
-		Hook:        hookName,
-		Optional:    true,
-		IgnoreError: ignoreError,
+func (m *RegistryManager) doCommitTransaction(t *state.Task, _ *tomb.Tomb) (err error) {
+	st := t.State()
+	st.Lock()
+	defer st.Unlock()
+
+	tx, _, err := GetStoredTransaction(t)
+	if err != nil {
+		return err
 	}
-	summary := fmt.Sprintf(i18n.G("Run hook %s of snap %q"), hookName, snapName)
-	task := hookstate.HookTask(st, summary, hookSup, nil)
-	return task
+
+	registryAssert, err := assertstateRegistry(st, tx.RegistryAccount, tx.RegistryName)
+	if err != nil {
+		return err
+	}
+	schema := registryAssert.Registry().Schema
+
+	return tx.Commit(st, schema)
+}
+
+func (m *RegistryManager) clearOngoingTransaction(t *state.Task, _ *tomb.Tomb) error {
+	st := t.State()
+	st.Lock()
+	defer st.Unlock()
+
+	tx, _, err := GetStoredTransaction(t)
+	if err != nil {
+		return err
+	}
+
+	err = unsetOngoingTransaction(st, tx.RegistryAccount, tx.RegistryName)
+	if err != nil {
+		return err
+	}
+
+	// TODO: unblock next waiting registry writer once we add the blocking logic
+	return nil
+}
+
+func setOngoingTransaction(st *state.State, account, registryName, commitTaskID string) error {
+	var commitTasks map[string]string
+	err := st.Get("registry-commit-tasks", &commitTasks)
+	if err != nil {
+		if !errors.Is(err, &state.NoStateError{}) {
+			return err
+		}
+
+		commitTasks = make(map[string]string, 1)
+	}
+
+	registryRef := account + "/" + registryName
+	if taskID, ok := commitTasks[registryRef]; ok {
+		return fmt.Errorf("internal error: cannot set task %q as ongoing commit task for registry %s: already have %q", commitTaskID, registryRef, taskID)
+	}
+
+	commitTasks[registryRef] = commitTaskID
+	st.Set("registry-commit-tasks", commitTasks)
+	return nil
 }
 
+func unsetOngoingTransaction(st *state.State, account, registryName string) error {
+	var commitTasks map[string]string
+	err := st.Get("registry-commit-tasks", &commitTasks)
+	if err != nil {
+		if errors.Is(err, &state.NoStateError{}) {
+			// already unset, nothing to do
+			return nil
+		}
+		return err
+	}
+
+	registryRef := account + "/" + registryName
+	if _, ok := commitTasks[registryRef]; !ok {
+		// already unset, nothing to do
+		return nil
+	}
+
+	delete(commitTasks, registryRef)
+
+	if len(commitTasks) == 0 {
+		st.Set("registry-commit-tasks", nil)
+	} else {
+		st.Set("registry-commit-tasks", commitTasks)
+	}
+
+	return nil
+}
+
+func (m *RegistryManager) noop(*state.Task, *tomb.Tomb) error { return nil }
+
 type changeViewHandler struct {
 	ctx *hookstate.Context
 }

overlord/registrystate/registrymgr_test.go

@@ -20,6 +20,7 @@ package registrystate_test
 
 import (
 	"errors"
+	"strings"
 	"time"
 
 	"github.com/snapcore/snapd/dirs"
@@ -356,3 +357,158 @@ func (s *registryTestSuite) TestManagerOk(c *C) {
 	err = s.o.Settle(5 * time.Second)
 	c.Assert(err, IsNil)
 }
+
+func (s *registryTestSuite) TestSetAndUnsetOngoingTransactionHelpers(c *C) {
+	s.state.Lock()
+	defer s.state.Unlock()
+
+	var commitTasks map[string]string
+	err := s.state.Get("registry-commit-tasks", &commitTasks)
+	c.Assert(err, testutil.ErrorIs, &state.NoStateError{})
+
+	err = registrystate.SetOngoingTransaction(s.state, "my-acc", "my-reg", "1")
+	c.Assert(err, IsNil)
+
+	// can't overwrite an ongoing commit task, since that could hide errors
+	err = registrystate.SetOngoingTransaction(s.state, "my-acc", "my-reg", "3")
+	c.Assert(err, ErrorMatches, `internal error: cannot set task "3" as ongoing commit task for registry my-acc/my-reg: already have "1"`)
+
+	err = registrystate.SetOngoingTransaction(s.state, "other-acc", "other-reg", "2")
+	c.Assert(err, IsNil)
+
+	err = s.state.Get("registry-commit-tasks", &commitTasks)
+	c.Assert(err, IsNil)
+	c.Assert(commitTasks["my-acc/my-reg"], Equals, "1")
+
+	err = registrystate.UnsetOngoingTransaction(s.state, "my-acc", "my-reg")
+	c.Assert(err, IsNil)
+
+	// unsetting non-existing key is fine
+	err = registrystate.UnsetOngoingTransaction(s.state, "my-acc", "my-reg")
+	c.Assert(err, IsNil)
+
+	err = s.state.Get("registry-commit-tasks", &commitTasks)
+	c.Assert(err, IsNil)
+	c.Assert(commitTasks["other-acc/other-reg"], Equals, "2")
+
+	err = registrystate.UnsetOngoingTransaction(s.state, "other-acc", "other-reg")
+	c.Assert(err, IsNil)
+
+	err = s.state.Get("registry-commit-tasks", &commitTasks)
+	c.Assert(err, testutil.ErrorIs, &state.NoStateError{})
+
+	// unsetting non-existing key is still fine when there's no map at all
+	err = registrystate.UnsetOngoingTransaction(s.state, "my-acc", "my-reg")
+	c.Assert(err, IsNil)
+}
+
+func (s *registryTestSuite) TestCommitTransaction(c *C) {
+	s.state.Lock()
+	defer s.state.Unlock()
+
+	chg := s.state.NewChange("test", "")
+	t := s.state.NewTask("commit-registry-tx", "")
+	chg.AddTask(t)
+
+	// attach a transaction with some changes to the commit task
+	tx, err := registrystate.NewTransaction(s.state, s.devAccID, "network")
+	c.Assert(err, IsNil)
+
+	err = tx.Set("wifi.ssid", "foo")
+	c.Assert(err, IsNil)
+
+	registrystate.SetTransaction(t, tx)
+
+	s.state.Unlock()
+	err = s.o.Settle(testutil.HostScaledTimeout(5 * time.Second))
+	s.state.Lock()
+	c.Assert(err, IsNil)
+
+	c.Assert(t.Status(), Equals, state.DoneStatus, Commentf(strings.Join(t.Log(), "\n")))
+
+	tx, _, err = registrystate.GetStoredTransaction(t)
+	c.Assert(err, IsNil)
+
+	// clearing would remove non-committed changes, so if we read the set value
+	// it's because it has been successfully committed
+	err = tx.Clear(s.state)
+	c.Assert(err, IsNil)
+
+	val, err := tx.Get("wifi.ssid")
+	c.Assert(err, IsNil)
+	c.Assert(val, Equals, "foo")
+}
+
+func (s *registryTestSuite) TestClearOngoingTransaction(c *C) {
+	s.state.Lock()
+	defer s.state.Unlock()
+
+	chg := s.state.NewChange("test", "")
+	commitTask := s.state.NewTask("commit-registry-tx", "")
+	commitTask.SetStatus(state.DoneStatus)
+	chg.AddTask(commitTask)
+
+	tx, err := registrystate.NewTransaction(s.state, s.devAccID, "network")
+	c.Assert(err, IsNil)
+	registrystate.SetTransaction(commitTask, tx)
+
+	t := s.state.NewTask("clear-registry-tx", "")
+	chg.AddTask(t)
+	t.Set("commit-task", commitTask.ID())
+
+	registrystate.SetOngoingTransaction(s.state, s.devAccID, "network", commitTask.ID())
+
+	var commitTasks map[string]string
+	err = s.state.Get("registry-commit-tasks", &commitTasks)
+	c.Assert(err, IsNil)
+
+	s.state.Unlock()
+	err = s.o.Settle(testutil.HostScaledTimeout(5 * time.Second))
+	s.state.Lock()
+	c.Assert(err, IsNil)
+	c.Assert(t.Status(), Equals, state.DoneStatus, Commentf(strings.Join(t.Log(), "\n")))
+
+	commitTasks = nil
+	err = s.state.Get("registry-commit-tasks", &commitTasks)
+	c.Assert(err, testutil.ErrorIs, &state.NoStateError{})
+}
+
+func (s *registryTestSuite) TestClearTransactionOnError(c *C) {
+	s.state.Lock()
+	defer s.state.Unlock()
+
+	chg := s.state.NewChange("test", "")
+	clearTask := s.state.NewTask("clear-registry-tx-on-error", "")
+	chg.AddTask(clearTask)
+
+	commitTask := s.state.NewTask("commit-registry-tx", "")
+	chg.AddTask(commitTask)
+	commitTask.WaitFor(clearTask)
+	clearTask.Set("commit-task", commitTask.ID())
+
+	tx, err := registrystate.NewTransaction(s.state, s.devAccID, "network")
+	c.Assert(err, IsNil)
+
+	// the schema will reject this, so the commit will fail
+	err = tx.Set("foo", "bar")
+	c.Assert(err, IsNil)
+	registrystate.SetTransaction(commitTask, tx)
+
+	// add this transaction to the state
+	registrystate.SetOngoingTransaction(s.state, s.devAccID, "network", commitTask.ID())
+
+	s.state.Unlock()
+	err = s.o.Settle(testutil.HostScaledTimeout(5 * time.Second))
+	s.state.Lock()
+	c.Assert(err, IsNil)
+
+	c.Assert(chg.Status(), Equals, state.ErrorStatus)
+	c.Assert(commitTask.Status(), Equals, state.ErrorStatus)
+	c.Assert(clearTask.Status(), Equals, state.UndoneStatus)
+	c.Assert(strings.Join(commitTask.Log(), "\n"), Matches, ".*ERROR cannot accept top level element: map contains unexpected key \"foo\"")
+
+	// no ongoing registry transaction
+	var commitTasks map[string]string
+	err = s.state.Get("registry-commit-tasks", &commitTasks)
+	c.Assert(err, testutil.ErrorIs, &state.NoStateError{})
+}

overlord/registrystate/registrystate_test.go

@@ -67,6 +67,17 @@ func (s *registryTestSuite) SetUpTest(c *C) {
 	s.state.Lock()
 	defer s.state.Unlock()
 
+	runner := s.o.TaskRunner()
+	s.o.AddManager(runner)
+
+	hookMgr, err := hookstate.Manager(s.state, runner)
+	c.Assert(err, IsNil)
+	s.o.AddManager(hookMgr)
+
+	// to test the registryManager
+	mgr := registrystate.Manager(s.state, hookMgr, runner)
+	s.o.AddManager(mgr)
+
 	storeSigning := assertstest.NewStoreStack("can0nical", nil)
 	db, err := asserts.OpenDatabase(&asserts.DatabaseConfig{
 		Backstore: asserts.NewMemoryBackstore(),