From 21896f39aa9203dad8000067005ebf8981f76ae9 Mon Sep 17 00:00:00 2001 From: Stephen Mwangi Date: Tue, 23 Jun 2026 23:05:24 +0300 Subject: [PATCH 1/7] o/devicemgmtstate: add task to queue response messages --- overlord/devicemgmtstate/devicemgmtmgr.go | 101 +++- .../devicemgmtstate/devicemgmtmgr_test.go | 448 +++++++++++++++++- 2 files changed, 544 insertions(+), 5 deletions(-) diff --git a/overlord/devicemgmtstate/devicemgmtmgr.go b/overlord/devicemgmtstate/devicemgmtmgr.go index 71b79756970..251b478caba 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr.go +++ b/overlord/devicemgmtstate/devicemgmtmgr.go @@ -25,6 +25,7 @@ package devicemgmtstate import ( + "encoding/json" "errors" "fmt" "sort" @@ -58,6 +59,8 @@ var ( maxSequences = 256 maxBlockedMessagesPerSequence = 8 + awaitSubsystemRetryInterval = 30 * time.Second + deviceMgmtExchangeChangeKind = swfeats.RegisterChangeKind("device-management-exchange") ) @@ -180,6 +183,23 @@ func (ms *deviceMgmtState) getRequestMessage(id string) (*RequestMessage, error) return nil, fmt.Errorf("cannot find message %q", id) } +// removeRequestMessage removes a processed request message from its sequence, +// leaving the sequence entry in place so its Applied progress is preserved for +// later messages in the same sequence. +func (ms *deviceMgmtState) removeRequestMessage(msg *RequestMessage) { + seq := ms.Sequences[msg.BaseID] + if seq == nil { + return + } + + for i, m := range seq.Messages { + if m.SeqNum == msg.SeqNum { + seq.Messages = append(seq.Messages[:i], seq.Messages[i+1:]...) + return + } + } +} + // enqueueRequestMessages queues incoming request messages for processing // and updates polling state accordingly. func (ms *deviceMgmtState) enqueueRequestMessages(pollResp *store.MessageExchangeResponse) { @@ -586,9 +606,86 @@ func (m *DeviceMgmtManager) doApplyMessage(t *state.Task, _ *tomb.Tomb) error { } // doQueueResponse builds a response, signs it, and queues it for transmission on the next exchange. -// Retries until subsystem change completes. +// Retries until the subsystem change (if any) completes. func (m *DeviceMgmtManager) doQueueResponse(t *state.Task, _ *tomb.Tomb) error { - // TODO: implement this task, no-op for now. + m.state.Lock() + defer m.state.Unlock() + + ms, err := m.getState() + if err != nil { + return err + } + + var msgID string + err = t.Get("message-id", &msgID) + if err != nil { + return err + } + + msg, err := ms.getRequestMessage(msgID) + if err != nil { + // Message already processed on a prior run. + return nil + } + + err = m.setMessageResponseFromChange(msg) + if err != nil { + return err + } + + bodyBytes, err := json.Marshal(msg.ResponseBody) + if err != nil { + return fmt.Errorf("cannot marshal response body: %w", err) + } + + resAs, err := m.signer.SignResponseMessage(msg.AccountID, msg.ID(), msg.ResponseStatus, bodyBytes) + if err != nil { + return fmt.Errorf("cannot sign response message: %w", err) + } + + ms.ReadyResponses[msg.ID()] = store.Message{ + Format: "assertion", + Data: string(asserts.Encode(resAs)), + } + + if msg.SeqNum > 0 { + ms.Sequences[msg.BaseID].Applied = msg.SeqNum + } + ms.removeRequestMessage(msg) + + m.setState(ms) + + return nil +} + +// setMessageResponseFromChange populates msg's response fields from the completed apply change. +func (m *DeviceMgmtManager) setMessageResponseFromChange(msg *RequestMessage) error { + if msg.ResponseStatus != "" { + return nil + } + + handler, ok := m.handlers[msg.Kind] + if !ok { + return fmt.Errorf("cannot find handler for message kind %q", msg.Kind) + } + + change := m.state.Change(msg.ApplyChangeID) + if change == nil { + return fmt.Errorf("cannot find subsystem change %q", msg.ApplyChangeID) + } + if !change.Status().Ready() { + return &state.Retry{After: awaitSubsystemRetryInterval} + } + + body, err := handler.ResultFromChange(change) + if err != nil { + msg.ResponseStatus = asserts.MessageStatusError + msg.ResponseBody = map[string]any{"message": err.Error()} + } else { + msg.ResponseStatus = asserts.MessageStatusSuccess + msg.ResponseBody = body + } + return nil } diff --git a/overlord/devicemgmtstate/devicemgmtmgr_test.go b/overlord/devicemgmtstate/devicemgmtmgr_test.go index 6bc470a8087..52fc444c6b7 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr_test.go +++ b/overlord/devicemgmtstate/devicemgmtmgr_test.go @@ -67,15 +67,35 @@ type mockMessageHandler struct { } func (h *mockMessageHandler) Validate(st *state.State, msg *devicemgmtstate.RequestMessage) error { - return h.validate(st, msg) + if h.validate != nil { + return h.validate(st, msg) + } + + return nil } func (h *mockMessageHandler) Apply(st *state.State, msg *devicemgmtstate.RequestMessage) (string, error) { - return h.apply(st, msg) + if h.apply != nil { + return h.apply(st, msg) + } + + return "", nil } func (h *mockMessageHandler) ResultFromChange(chg *state.Change) (map[string]any, error) { - return h.resultFromChange(chg) + if h.resultFromChange != nil { + return h.resultFromChange(chg) + } + + return nil, nil +} + +type mockSigner struct { + sign func(accountID, messageID string, status asserts.MessageStatus, body []byte) (*asserts.ResponseMessage, error) +} + +func (s *mockSigner) SignResponseMessage(accountID, messageID string, status asserts.MessageStatus, body []byte) (*asserts.ResponseMessage, error) { + return s.sign(accountID, messageID, status, body) } func setRemoteMgmtFeatureFlag(c *C, st *state.State, value any) { @@ -403,6 +423,8 @@ func (s *deviceMgmtMgrSuite) TestDoExchangeMessagesFetchOK(c *C) { }, nil }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.settle(c) ms, err := s.mgr.GetState() @@ -437,6 +459,8 @@ func (s *deviceMgmtMgrSuite) TestDoExchangeMessagesReplyOK(c *C) { }, nil }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + ms := &devicemgmtstate.DeviceMgmtState{ Sequences: make(map[string]*devicemgmtstate.SequenceState), LastReceivedToken: "token-123", @@ -473,6 +497,8 @@ func (s *deviceMgmtMgrSuite) TestDoExchangeMessagesSequenceLRU(c *C) { }, nil }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.settle(c) ms, err := s.mgr.GetState() @@ -501,6 +527,8 @@ func (s *deviceMgmtMgrSuite) TestDoExchangeMessagesInvalidMessage(c *C) { }, nil }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.settle(c) c.Check(s.logbuf.String(), testutil.Contains, "cannot parse request-message with token token-123") @@ -526,6 +554,8 @@ func (s *deviceMgmtMgrSuite) TestDoExchangeMessagesDuplicateMessage(c *C) { }, nil }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.settle(c) ms, err := s.mgr.GetState() @@ -586,6 +616,8 @@ func (s *deviceMgmtMgrSuite) TestDoExchangeMessagesIdempotent(c *C) { }, nil }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.settle(c) ms, err := s.mgr.GetState() @@ -615,6 +647,8 @@ func (s *deviceMgmtMgrSuite) TestDoDispatchMessagesUnsequenced(c *C) { }, nil }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.settle(c) // Exchange 2: msg1 is dedup'd by exchange; msg2 and msg3 are new. @@ -870,6 +904,8 @@ func (s *deviceMgmtMgrSuite) TestDoDispatchMessagesEvictedSequenceRejected(c *C) return &store.MessageExchangeResponse{Messages: messages}, nil }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.settle(c) ms, err := s.mgr.GetState() @@ -915,6 +951,8 @@ func (s *deviceMgmtMgrSuite) TestDoDispatchMessagesBlockedSequenceRejected(c *C) return &store.MessageExchangeResponse{Messages: messages}, nil }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.settle(c) ms, err := s.mgr.GetState() @@ -941,6 +979,8 @@ func (s *deviceMgmtMgrSuite) TestDoDispatchMessagesIdempotent(c *C) { return &store.MessageExchangeResponse{}, nil }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + ms := &devicemgmtstate.DeviceMgmtState{ Sequences: map[string]*devicemgmtstate.SequenceState{ "msg1": { @@ -1010,6 +1050,8 @@ func (s *deviceMgmtMgrSuite) TestDoApplyMessageOK(c *C) { }, nil }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.settle(c) ms, err := s.mgr.GetState() @@ -1060,6 +1102,8 @@ func (s *deviceMgmtMgrSuite) TestDoApplyMessageSkipIfAlreadyFailed(c *C) { }, }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.settle(c) ms, err := s.mgr.GetState() @@ -1082,6 +1126,8 @@ func (s *deviceMgmtMgrSuite) TestDoApplyMessageNoHandlerForMessageKind(c *C) { }, nil }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.settle(c) ms, err := s.mgr.GetState() @@ -1111,6 +1157,8 @@ func (s *deviceMgmtMgrSuite) TestDoApplyMessageApplyError(c *C) { }, }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.settle(c) ms, err := s.mgr.GetState() @@ -1140,6 +1188,8 @@ func (s *deviceMgmtMgrSuite) TestDoApplyMessageIdempotent(c *C) { }, }) + s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + ms := &devicemgmtstate.DeviceMgmtState{ Sequences: map[string]*devicemgmtstate.SequenceState{ "msg1": { @@ -1284,6 +1334,398 @@ func (s *deviceMgmtMgrSuite) TestDoApplyMessageMessageNotFound(c *C) { c.Assert(err, ErrorMatches, `cannot find message "seqA-2"`) } +func (s *deviceMgmtMgrSuite) TestDoQueueResponseOK(c *C) { + s.st.Lock() + defer s.st.Unlock() + + s.mockStore(func(_ context.Context, _ *store.MessageExchangeRequest) (*store.MessageExchangeResponse, error) { + return &store.MessageExchangeResponse{ + Messages: []store.MessageWithToken{ + s.makeStoreRequestMessage(c, "mesg-1", "test-kind", "token-1"), + }, + }, nil + }) + + s.mgr.RegisterHandler("test-kind", &mockMessageHandler{ + validate: func(*state.State, *devicemgmtstate.RequestMessage) error { return nil }, + apply: func(st *state.State, msg *devicemgmtstate.RequestMessage) (string, error) { + chg := st.NewChange("subsystem", "apply payload") + devicemgmtstate.MarkChangeForMessage(chg, msg) + return chg.ID(), nil + }, + resultFromChange: func(*state.Change) (map[string]any, error) { + return map[string]any{"values": "ok"}, nil + }, + }) + + s.mgr.MockSigner(&mockSigner{ + sign: func(accountID, messageID string, status asserts.MessageStatus, body []byte) (*asserts.ResponseMessage, error) { + c.Check(accountID, Equals, "my-brand") + c.Check(messageID, Equals, "mesg-1") + c.Check(status, Equals, asserts.MessageStatusSuccess) + c.Check(string(body), Equals, `{"values":"ok"}`) + + return assertstest.FakeAssertionWithBody(body, map[string]any{ + "type": "response-message", + "account-id": accountID, + "message-id": messageID, + "device": "serial-1.my-model.my-brand", + "status": string(status), + "body-length": strconv.Itoa(len(body)), + }).(*asserts.ResponseMessage), nil + }, + }) + + s.settle(c) + + ms, err := s.mgr.GetState() + c.Assert(err, IsNil) + + c.Check(ms.Sequences["mesg"].Messages, HasLen, 0) + c.Assert(ms.ReadyResponses, HasLen, 1) + c.Check(ms.ReadyResponses["mesg-1"].Format, Equals, "assertion") + c.Check(ms.Sequences["mesg"].Applied, Equals, 1) +} + +func (s *deviceMgmtMgrSuite) TestDoQueueResponseStatusAlreadyKnown(c *C) { + s.st.Lock() + defer s.st.Unlock() + + s.mockStore(func(_ context.Context, _ *store.MessageExchangeRequest) (*store.MessageExchangeResponse, error) { + return &store.MessageExchangeResponse{ + Messages: []store.MessageWithToken{ + s.makeStoreRequestMessage(c, "mesg-1", "test-kind", "token-1"), + }, + }, nil + }) + + s.runner.AddHandler("validate-mgmt-message", func(t *state.Task, _ *tomb.Tomb) error { + t.State().Lock() + defer t.State().Unlock() + + ms, err := s.mgr.GetState() + c.Assert(err, IsNil) + + ms.Sequences["mesg"].Messages[0].ResponseStatus = asserts.MessageStatusRejected + ms.Sequences["mesg"].Messages[0].ResponseBody = map[string]any{"message": "device not in target list"} + s.mgr.SetState(ms) + + return nil + }, nil) + + s.mgr.RegisterHandler("test-kind", &mockMessageHandler{ + apply: func(*state.State, *devicemgmtstate.RequestMessage) (string, error) { + c.Fatal("apply must not be called when ResponseStatus is already set") + + return "", nil + }, + resultFromChange: func(*state.Change) (map[string]any, error) { + // A message whose ResponseStatus was set earlier in the pipeline (e.g. by + // rejectSequence) must be signed and queued without calling handler.ResultFromChange. + c.Fatal("resultFromChange must not be called when ResponseStatus is already set") + + return nil, nil + }, + }) + + s.mgr.MockSigner(&mockSigner{ + sign: func(accountID, messageID string, status asserts.MessageStatus, body []byte) (*asserts.ResponseMessage, error) { + c.Check(messageID, Equals, "mesg-1") + c.Check(status, Equals, asserts.MessageStatusRejected) + c.Check(string(body), Equals, `{"message":"device not in target list"}`) + + return assertstest.FakeAssertionWithBody(body, map[string]any{ + "type": "response-message", + "account-id": accountID, + "message-id": messageID, + "device": "serial-1.my-model.my-brand", + "status": string(status), + "body-length": strconv.Itoa(len(body)), + }).(*asserts.ResponseMessage), nil + }, + }) + + s.settle(c) + + ms, err := s.mgr.GetState() + c.Assert(err, IsNil) + + c.Check(ms.Sequences["mesg"].Messages, HasLen, 0) + c.Assert(ms.ReadyResponses, HasLen, 1) +} + +func (s *deviceMgmtMgrSuite) TestDoQueueResponseIdempotent(c *C) { + s.st.Lock() + defer s.st.Unlock() + + signCalls := 0 + s.mgr.MockSigner(&mockSigner{ + sign: func(accountID, messageID string, status asserts.MessageStatus, body []byte) (*asserts.ResponseMessage, error) { + signCalls++ + return assertstest.FakeAssertionWithBody(body, map[string]any{ + "type": "response-message", + "account-id": accountID, + "message-id": messageID, + "device": "serial-1.my-model.my-brand", + "status": string(status), + "body-length": strconv.Itoa(len(body)), + }).(*asserts.ResponseMessage), nil + }, + }) + + sentResponses := 0 + s.mockStore(func(_ context.Context, req *store.MessageExchangeRequest) (*store.MessageExchangeResponse, error) { + sentResponses += len(req.Messages) + return &store.MessageExchangeResponse{}, nil + }) + + ms := &devicemgmtstate.DeviceMgmtState{ + Sequences: map[string]*devicemgmtstate.SequenceState{ + "msg1": { + Messages: []*devicemgmtstate.RequestMessage{ + { + AccountID: "my-brand", + AuthorityID: "my-brand", + BaseID: "msg1", + Kind: "test-kind", + Devices: []string{"serial-1.my-model.my-brand"}, + ValidSince: fixedTestTime, + ValidUntil: fixedTestTime.Add(24 * time.Hour), + Body: `{"action": "get", "account": "my-brand", "view": "network/wifi-state"}`, + ResponseStatus: asserts.MessageStatusSuccess, + ResponseBody: map[string]any{"values": "ok"}, + }, + }, + }, + }, + ReadyResponses: make(map[string]store.Message), + } + s.mgr.SetState(ms) + + chg := s.st.NewChange("test", "test change") + for i := 1; i <= 3; i++ { + t := s.st.NewTask("queue-mgmt-response", fmt.Sprintf("queue msg1 attempt %d", i)) + t.Set("message-id", "msg1") + chg.AddTask(t) + } + + s.settle(c) + + c.Check(chg.Status(), Equals, state.DoneStatus) + c.Check(signCalls, Equals, 1) + c.Check(sentResponses, Equals, 1) +} + +func (s *deviceMgmtMgrSuite) TestDoQueueResponseResultFromChangeError(c *C) { + s.st.Lock() + defer s.st.Unlock() + + s.mockStore(func(_ context.Context, _ *store.MessageExchangeRequest) (*store.MessageExchangeResponse, error) { + return &store.MessageExchangeResponse{ + Messages: []store.MessageWithToken{ + s.makeStoreRequestMessage(c, "mesg-1", "test-kind", "token-1"), + }, + }, nil + }) + + s.mgr.RegisterHandler("test-kind", &mockMessageHandler{ + validate: func(*state.State, *devicemgmtstate.RequestMessage) error { return nil }, + apply: func(st *state.State, msg *devicemgmtstate.RequestMessage) (string, error) { + chg := st.NewChange("subsystem", "apply payload") + devicemgmtstate.MarkChangeForMessage(chg, msg) + return chg.ID(), nil + }, + resultFromChange: func(*state.Change) (map[string]any, error) { + return nil, fmt.Errorf("operation failed") + }, + }) + + s.mgr.MockSigner(&mockSigner{ + sign: func(accountID, messageID string, status asserts.MessageStatus, body []byte) (*asserts.ResponseMessage, error) { + c.Check(messageID, Equals, "mesg-1") + c.Check(status, Equals, asserts.MessageStatusError) + c.Check(string(body), Equals, `{"message":"operation failed"}`) + + return assertstest.FakeAssertionWithBody(body, map[string]any{ + "type": "response-message", + "account-id": accountID, + "message-id": messageID, + "device": "serial-1.my-model.my-brand", + "status": string(status), + "body-length": strconv.Itoa(len(body)), + }).(*asserts.ResponseMessage), nil + }, + }) + + s.settle(c) + + ms, err := s.mgr.GetState() + c.Assert(err, IsNil) + + c.Check(ms.Sequences["mesg"].Messages, HasLen, 0) + c.Assert(ms.ReadyResponses, HasLen, 1) + c.Check(ms.ReadyResponses["mesg-1"].Format, Equals, "assertion") +} + +func (s *deviceMgmtMgrSuite) TestDoQueueResponseSubsystemChangeNotFound(c *C) { + s.st.Lock() + defer s.st.Unlock() + + ms := &devicemgmtstate.DeviceMgmtState{ + Sequences: map[string]*devicemgmtstate.SequenceState{ + "msg1": { + Messages: []*devicemgmtstate.RequestMessage{ + { + AccountID: "my-brand", + AuthorityID: "my-brand", + BaseID: "msg1", + Kind: "test-kind", + Devices: []string{"serial-1.my-model.my-brand"}, + ValidSince: fixedTestTime, + ValidUntil: fixedTestTime.Add(24 * time.Hour), + Body: `{"action": "get", "account": "my-brand", "view": "network/wifi-state"}`, + ApplyChangeID: "16384", + }, + }, + }, + }, + ReadyResponses: make(map[string]store.Message), + } + s.mgr.SetState(ms) + + s.mgr.RegisterHandler("test-kind", &mockMessageHandler{ + resultFromChange: func(*state.Change) (map[string]any, error) { + c.Error("resultFromChange must not be called when subsystem change cannot be found") + + return nil, nil + }, + }) + + chg := s.st.NewChange("test", "test change") + t := s.st.NewTask("queue-mgmt-response", "queue response for msg1") + t.Set("message-id", "msg1") + chg.AddTask(t) + + s.st.Unlock() + err := s.mgr.DoQueueResponse(t, &tomb.Tomb{}) + s.st.Lock() + c.Assert(err, ErrorMatches, `cannot find subsystem change "16384"`) +} + +func (s *deviceMgmtMgrSuite) TestDoQueueResponseSubsystemChangeNotReady(c *C) { + s.st.Lock() + defer s.st.Unlock() + + subsysChg := s.st.NewChange("subsys-op", "subsystem operation") + subsysChg.SetStatus(state.DoingStatus) + + ms := &devicemgmtstate.DeviceMgmtState{ + Sequences: map[string]*devicemgmtstate.SequenceState{ + "msg1": { + Messages: []*devicemgmtstate.RequestMessage{ + { + AccountID: "my-brand", + AuthorityID: "my-brand", + BaseID: "msg1", + Kind: "test-kind", + Devices: []string{"serial-1.my-model.my-brand"}, + ValidSince: fixedTestTime, + ValidUntil: fixedTestTime.Add(24 * time.Hour), + Body: `{"action": "get", "account": "my-brand", "view": "network/wifi-state"}`, + ApplyChangeID: subsysChg.ID(), + }, + }, + }, + }, + ReadyResponses: make(map[string]store.Message), + } + s.mgr.SetState(ms) + + s.mgr.RegisterHandler("test-kind", &mockMessageHandler{ + resultFromChange: func(*state.Change) (map[string]any, error) { + c.Error("resultFromChange must not be called while subsystem change is not ready") + + return nil, nil + }, + }) + + chg := s.st.NewChange("test", "test change") + t := s.st.NewTask("queue-mgmt-response", "queue response for msg1") + t.Set("message-id", "msg1") + chg.AddTask(t) + + s.st.Unlock() + err := s.mgr.DoQueueResponse(t, &tomb.Tomb{}) + s.st.Lock() + c.Assert(err, FitsTypeOf, &state.Retry{}) +} + +func (s *deviceMgmtMgrSuite) TestDoQueueResponseMessageNotFound(c *C) { + s.st.Lock() + defer s.st.Unlock() + + ms := &devicemgmtstate.DeviceMgmtState{ + Sequences: make(map[string]*devicemgmtstate.SequenceState), + ReadyResponses: make(map[string]store.Message), + } + s.mgr.SetState(ms) + + chg := s.st.NewChange("test", "test change") + t := s.st.NewTask("queue-mgmt-response", "queue response for unknown message") + t.Set("message-id", "msg1") + chg.AddTask(t) + + s.st.Unlock() + err := s.mgr.DoQueueResponse(t, &tomb.Tomb{}) + s.st.Lock() + // Message not found means it was already processed on a prior run; handler returns nil. + c.Assert(err, IsNil) +} + +func (s *deviceMgmtMgrSuite) TestDoQueueResponseSigningError(c *C) { + s.st.Lock() + defer s.st.Unlock() + + s.mockStore(func(_ context.Context, _ *store.MessageExchangeRequest) (*store.MessageExchangeResponse, error) { + return &store.MessageExchangeResponse{ + Messages: []store.MessageWithToken{ + s.makeStoreRequestMessage(c, "mesg-1", "test-kind", "token-1"), + }, + }, nil + }) + + s.mgr.RegisterHandler("test-kind", &mockMessageHandler{ + validate: func(*state.State, *devicemgmtstate.RequestMessage) error { return nil }, + apply: func(st *state.State, msg *devicemgmtstate.RequestMessage) (string, error) { + chg := st.NewChange("subsystem", "apply payload") + devicemgmtstate.MarkChangeForMessage(chg, msg) + return chg.ID(), nil + }, + resultFromChange: func(*state.Change) (map[string]any, error) { + return map[string]any{"values": "ok"}, nil + }, + }) + + s.mgr.MockSigner(&mockSigner{ + sign: func(_, _ string, _ asserts.MessageStatus, _ []byte) (*asserts.ResponseMessage, error) { + return nil, fmt.Errorf("device key not found") + }, + }) + + s.settle(c) + + var found bool + for _, chg := range s.st.Changes() { + for _, t := range chg.Tasks() { + if t.Kind() == "queue-mgmt-response" && t.Status() == state.ErrorStatus { + c.Check(strings.Join(t.Log(), "\n"), testutil.Contains, "cannot sign response message: device key not found") + found = true + } + } + } + c.Assert(found, Equals, true) +} + func (s *deviceMgmtMgrSuite) TestParseRequestMessageInvalid(c *C) { type test struct { name string From d7e0cf090437fed78f6a7a9010dd1b2f2c341a60 Mon Sep 17 00:00:00 2001 From: Stephen Mwangi Date: Wed, 24 Jun 2026 13:39:13 +0300 Subject: [PATCH 2/7] o/devicemgmtstate: test that a task chain error does not affect other messages --- overlord/devicemgmtstate/devicemgmtmgr.go | 2 - .../devicemgmtstate/devicemgmtmgr_test.go | 116 +++++++++++++----- 2 files changed, 84 insertions(+), 34 deletions(-) diff --git a/overlord/devicemgmtstate/devicemgmtmgr.go b/overlord/devicemgmtstate/devicemgmtmgr.go index 251b478caba..cb2cc98ef5e 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr.go +++ b/overlord/devicemgmtstate/devicemgmtmgr.go @@ -499,8 +499,6 @@ func (m *DeviceMgmtManager) dispatchSequence(dispatchTask *state.Task, seq *sequ // the final task so callers can chain subsequent messages after it. func (m *DeviceMgmtManager) dispatchMessage(prevTask *state.Task, msg *RequestMessage) *state.Task { chg := prevTask.Change() - // TODO: add tests verifying that a failure in one message's task chain does not - // affect other messages (lanes provide this isolation, but it needs test coverage). lane := m.state.NewLane() addTask := func(kind, summary string) { diff --git a/overlord/devicemgmtstate/devicemgmtmgr_test.go b/overlord/devicemgmtstate/devicemgmtmgr_test.go index 52fc444c6b7..30a276ae717 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr_test.go +++ b/overlord/devicemgmtstate/devicemgmtmgr_test.go @@ -1038,6 +1038,82 @@ func (s *deviceMgmtMgrSuite) TestDoDispatchMessagesIdempotent(c *C) { c.Check(ti.queue["msg2"], NotNil) } +func (s *deviceMgmtMgrSuite) TestDispatchMessagesLaneIsolation(c *C) { + s.st.Lock() + defer s.st.Unlock() + + s.mockStore(func(_ context.Context, _ *store.MessageExchangeRequest) (*store.MessageExchangeResponse, error) { + return &store.MessageExchangeResponse{ + Messages: []store.MessageWithToken{ + s.makeStoreRequestMessage(c, "msg1", "test-kind", "token-1"), + s.makeStoreRequestMessage(c, "msg2", "test-kind", "token-2"), + }, + }, nil + }) + + // Override validate to simulate an internal error for msg1, + // causing the task runner to put msg1's chain in error/hold. + s.runner.AddHandler("validate-mgmt-message", func(t *state.Task, _ *tomb.Tomb) error { + t.State().Lock() + defer t.State().Unlock() + + var msgID string + err := t.Get("message-id", &msgID) + c.Assert(err, IsNil) + + if msgID == "msg1" { + return fmt.Errorf("internal error: unexpected state for msg1") + } + + return nil + }, nil) + + s.mgr.RegisterHandler("test-kind", &mockMessageHandler{ + apply: func(st *state.State, msg *devicemgmtstate.RequestMessage) (string, error) { + chg := st.NewChange("subsystem", "apply payload") + devicemgmtstate.MarkChangeForMessage(chg, msg) + return chg.ID(), nil + }, + resultFromChange: func(*state.Change) (map[string]any, error) { + return map[string]any{"result": "ok"}, nil + }, + }) + + s.mgr.MockSigner(&mockSigner{ + sign: func(accountID, messageID string, status asserts.MessageStatus, body []byte) (*asserts.ResponseMessage, error) { + return assertstest.FakeAssertionWithBody(body, map[string]any{ + "type": "response-message", + "account-id": accountID, + "message-id": messageID, + "device": "serial-1.my-model.my-brand", + "status": string(status), + "body-length": strconv.Itoa(len(body)), + }).(*asserts.ResponseMessage), nil + }, + }) + + s.settle(c) + + changes := changesOfKind(s.st.Changes(), "device-management-exchange") + c.Assert(changes, HasLen, 1) + ti := buildTaskIndex(changes[0]) + + // msg1's chain is held due to the validate task's error. + c.Check(ti.validate["msg1"].Status(), Equals, state.ErrorStatus) + c.Check(ti.apply["msg1"].Status(), Equals, state.HoldStatus) + c.Check(ti.queue["msg1"].Status(), Equals, state.HoldStatus) + + // msg2's chain completes independently. + c.Check(ti.validate["msg2"].Status(), Equals, state.DoneStatus) + c.Check(ti.apply["msg2"].Status(), Equals, state.DoneStatus) + c.Check(ti.queue["msg2"].Status(), Equals, state.DoneStatus) + + ms, err := s.mgr.GetState() + c.Assert(err, IsNil) + c.Assert(ms.ReadyResponses, HasLen, 1) + c.Check(ms.ReadyResponses["msg2"].Format, Equals, "assertion") +} + func (s *deviceMgmtMgrSuite) TestDoApplyMessageOK(c *C) { s.st.Lock() defer s.st.Unlock() @@ -1660,28 +1736,6 @@ func (s *deviceMgmtMgrSuite) TestDoQueueResponseSubsystemChangeNotReady(c *C) { c.Assert(err, FitsTypeOf, &state.Retry{}) } -func (s *deviceMgmtMgrSuite) TestDoQueueResponseMessageNotFound(c *C) { - s.st.Lock() - defer s.st.Unlock() - - ms := &devicemgmtstate.DeviceMgmtState{ - Sequences: make(map[string]*devicemgmtstate.SequenceState), - ReadyResponses: make(map[string]store.Message), - } - s.mgr.SetState(ms) - - chg := s.st.NewChange("test", "test change") - t := s.st.NewTask("queue-mgmt-response", "queue response for unknown message") - t.Set("message-id", "msg1") - chg.AddTask(t) - - s.st.Unlock() - err := s.mgr.DoQueueResponse(t, &tomb.Tomb{}) - s.st.Lock() - // Message not found means it was already processed on a prior run; handler returns nil. - c.Assert(err, IsNil) -} - func (s *deviceMgmtMgrSuite) TestDoQueueResponseSigningError(c *C) { s.st.Lock() defer s.st.Unlock() @@ -1714,16 +1768,14 @@ func (s *deviceMgmtMgrSuite) TestDoQueueResponseSigningError(c *C) { s.settle(c) - var found bool - for _, chg := range s.st.Changes() { - for _, t := range chg.Tasks() { - if t.Kind() == "queue-mgmt-response" && t.Status() == state.ErrorStatus { - c.Check(strings.Join(t.Log(), "\n"), testutil.Contains, "cannot sign response message: device key not found") - found = true - } - } - } - c.Assert(found, Equals, true) + changes := changesOfKind(s.st.Changes(), "device-management-exchange") + c.Assert(changes, HasLen, 1) + ti := buildTaskIndex(changes[0]) + + queueTask := ti.queue["mesg-1"] + c.Assert(queueTask, NotNil) + c.Check(queueTask.Status(), Equals, state.ErrorStatus) + c.Check(strings.Join(queueTask.Log(), "\n"), testutil.Contains, "cannot sign response message: device key not found") } func (s *deviceMgmtMgrSuite) TestParseRequestMessageInvalid(c *C) { From 7ea38cb3e2873bdd503d5f20639b153712427066 Mon Sep 17 00:00:00 2001 From: Stephen Mwangi Date: Wed, 24 Jun 2026 17:57:05 +0300 Subject: [PATCH 3/7] o/devicemgmtstate: set error message for unknown message kind in setMessageResponseFromChange --- overlord/devicemgmtstate/devicemgmtmgr.go | 4 +- .../devicemgmtstate/devicemgmtmgr_test.go | 61 +++++++++++++++++++ 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/overlord/devicemgmtstate/devicemgmtmgr.go b/overlord/devicemgmtstate/devicemgmtmgr.go index cb2cc98ef5e..96dca385875 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr.go +++ b/overlord/devicemgmtstate/devicemgmtmgr.go @@ -664,7 +664,9 @@ func (m *DeviceMgmtManager) setMessageResponseFromChange(msg *RequestMessage) er handler, ok := m.handlers[msg.Kind] if !ok { - return fmt.Errorf("cannot find handler for message kind %q", msg.Kind) + msg.ResponseStatus = asserts.MessageStatusError + msg.ResponseBody = map[string]any{"message": fmt.Sprintf("cannot find handler for message kind %q", msg.Kind)} + return nil } change := m.state.Change(msg.ApplyChangeID) diff --git a/overlord/devicemgmtstate/devicemgmtmgr_test.go b/overlord/devicemgmtstate/devicemgmtmgr_test.go index 30a276ae717..927a1b0605a 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr_test.go +++ b/overlord/devicemgmtstate/devicemgmtmgr_test.go @@ -1688,6 +1688,67 @@ func (s *deviceMgmtMgrSuite) TestDoQueueResponseSubsystemChangeNotFound(c *C) { c.Assert(err, ErrorMatches, `cannot find subsystem change "16384"`) } +func (s *deviceMgmtMgrSuite) TestDoQueueResponseNoHandlerForMessageKind(c *C) { + s.st.Lock() + defer s.st.Unlock() + + ms := &devicemgmtstate.DeviceMgmtState{ + Sequences: map[string]*devicemgmtstate.SequenceState{ + "msg1": { + Messages: []*devicemgmtstate.RequestMessage{ + { + AccountID: "my-brand", + AuthorityID: "my-brand", + BaseID: "msg1", + Kind: "unknown-kind", + Devices: []string{"serial-1.my-model.my-brand"}, + ValidSince: fixedTestTime, + ValidUntil: fixedTestTime.Add(24 * time.Hour), + Body: `what is this?`, + }, + }, + }, + }, + ReadyResponses: make(map[string]store.Message), + } + s.mgr.SetState(ms) + + s.mgr.MockSigner(&mockSigner{ + sign: func(accountID, messageID string, status asserts.MessageStatus, body []byte) (*asserts.ResponseMessage, error) { + c.Check(accountID, Equals, "my-brand") + c.Check(messageID, Equals, "msg1") + c.Check(status, Equals, asserts.MessageStatusError) + c.Check(string(body), Equals, `{"message":"cannot find handler for message kind \"unknown-kind\""}`) + + return assertstest.FakeAssertionWithBody(body, map[string]any{ + "type": "response-message", + "account-id": accountID, + "message-id": messageID, + "device": "serial-1.my-model.my-brand", + "status": string(status), + "body-length": strconv.Itoa(len(body)), + }).(*asserts.ResponseMessage), nil + }, + }) + + chg := s.st.NewChange("test", "test change") + t := s.st.NewTask("queue-mgmt-response", "queue response for msg1") + t.Set("message-id", "msg1") + chg.AddTask(t) + + s.st.Unlock() + err := s.mgr.DoQueueResponse(t, &tomb.Tomb{}) + s.st.Lock() + c.Assert(err, IsNil) + + ms, err = s.mgr.GetState() + c.Assert(err, IsNil) + + c.Check(ms.Sequences["msg1"].Messages, HasLen, 0) + c.Assert(ms.ReadyResponses, HasLen, 1) + c.Check(ms.ReadyResponses["msg1"].Format, Equals, "assertion") +} + func (s *deviceMgmtMgrSuite) TestDoQueueResponseSubsystemChangeNotReady(c *C) { s.st.Lock() defer s.st.Unlock() From 1fef18991afb00e6a0e1f679a302145e7053e04e Mon Sep 17 00:00:00 2001 From: Stephen Mwangi Date: Wed, 24 Jun 2026 18:33:44 +0300 Subject: [PATCH 4/7] o/devicemgmtstate: do not advance Applied on non-success response --- overlord/devicemgmtstate/devicemgmtmgr.go | 9 ++++++++- overlord/devicemgmtstate/devicemgmtmgr_test.go | 6 ++++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/overlord/devicemgmtstate/devicemgmtmgr.go b/overlord/devicemgmtstate/devicemgmtmgr.go index 96dca385875..508a921b87c 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr.go +++ b/overlord/devicemgmtstate/devicemgmtmgr.go @@ -646,7 +646,14 @@ func (m *DeviceMgmtManager) doQueueResponse(t *state.Task, _ *tomb.Tomb) error { Data: string(asserts.Encode(resAs)), } - if msg.SeqNum > 0 { + // TODO: rejecting sequences currently happens in 2 ways: + // 1. doDispatchMessage can evict the sequence immediately if it's rejected early. + // 2. If it errors elsewhere (in validate, apply, or queue-response), we end + // up not advancing Applied, which means we accumulate messages until we + // hit the sequence cap. + // Refactor sequence rejection to always evict immediately and determine the + // right behavior for internal errors. + if msg.SeqNum > 0 && msg.ResponseStatus == asserts.MessageStatusSuccess { ms.Sequences[msg.BaseID].Applied = msg.SeqNum } ms.removeRequestMessage(msg) diff --git a/overlord/devicemgmtstate/devicemgmtmgr_test.go b/overlord/devicemgmtstate/devicemgmtmgr_test.go index 927a1b0605a..69c04e2b180 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr_test.go +++ b/overlord/devicemgmtstate/devicemgmtmgr_test.go @@ -1528,6 +1528,7 @@ func (s *deviceMgmtMgrSuite) TestDoQueueResponseStatusAlreadyKnown(c *C) { c.Check(ms.Sequences["mesg"].Messages, HasLen, 0) c.Assert(ms.ReadyResponses, HasLen, 1) + c.Check(ms.Sequences["mesg"].Applied, Equals, 0) } func (s *deviceMgmtMgrSuite) TestDoQueueResponseIdempotent(c *C) { @@ -1641,6 +1642,7 @@ func (s *deviceMgmtMgrSuite) TestDoQueueResponseResultFromChangeError(c *C) { c.Check(ms.Sequences["mesg"].Messages, HasLen, 0) c.Assert(ms.ReadyResponses, HasLen, 1) c.Check(ms.ReadyResponses["mesg-1"].Format, Equals, "assertion") + c.Check(ms.Sequences["mesg"].Applied, Equals, 0) } func (s *deviceMgmtMgrSuite) TestDoQueueResponseSubsystemChangeNotFound(c *C) { @@ -1837,6 +1839,10 @@ func (s *deviceMgmtMgrSuite) TestDoQueueResponseSigningError(c *C) { c.Assert(queueTask, NotNil) c.Check(queueTask.Status(), Equals, state.ErrorStatus) c.Check(strings.Join(queueTask.Log(), "\n"), testutil.Contains, "cannot sign response message: device key not found") + + ms, err := s.mgr.GetState() + c.Assert(err, IsNil) + c.Check(ms.Sequences["mesg"].Applied, Equals, 0) } func (s *deviceMgmtMgrSuite) TestParseRequestMessageInvalid(c *C) { From 7652e2ab9728b55ad99e4c9fa964c9669169d5d8 Mon Sep 17 00:00:00 2001 From: Stephen Mwangi Date: Thu, 25 Jun 2026 11:47:16 +0300 Subject: [PATCH 5/7] o/devicemgmtstate: add reuseable noopTask in tests --- .../devicemgmtstate/devicemgmtmgr_test.go | 32 ++++++++++--------- 1 file changed, 17 insertions(+), 15 deletions(-) diff --git a/overlord/devicemgmtstate/devicemgmtmgr_test.go b/overlord/devicemgmtstate/devicemgmtmgr_test.go index 69c04e2b180..0b22ff053d3 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr_test.go +++ b/overlord/devicemgmtstate/devicemgmtmgr_test.go @@ -50,6 +50,8 @@ import ( func Test(t *testing.T) { TestingT(t) } +var noopTask = func(*state.Task, *tomb.Tomb) error { return nil } + type mockStore struct { storetest.Store @@ -423,7 +425,7 @@ func (s *deviceMgmtMgrSuite) TestDoExchangeMessagesFetchOK(c *C) { }, nil }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) s.settle(c) @@ -459,7 +461,7 @@ func (s *deviceMgmtMgrSuite) TestDoExchangeMessagesReplyOK(c *C) { }, nil }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) ms := &devicemgmtstate.DeviceMgmtState{ Sequences: make(map[string]*devicemgmtstate.SequenceState), @@ -497,7 +499,7 @@ func (s *deviceMgmtMgrSuite) TestDoExchangeMessagesSequenceLRU(c *C) { }, nil }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) s.settle(c) @@ -527,7 +529,7 @@ func (s *deviceMgmtMgrSuite) TestDoExchangeMessagesInvalidMessage(c *C) { }, nil }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) s.settle(c) @@ -554,7 +556,7 @@ func (s *deviceMgmtMgrSuite) TestDoExchangeMessagesDuplicateMessage(c *C) { }, nil }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) s.settle(c) @@ -616,7 +618,7 @@ func (s *deviceMgmtMgrSuite) TestDoExchangeMessagesIdempotent(c *C) { }, nil }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) s.settle(c) @@ -647,7 +649,7 @@ func (s *deviceMgmtMgrSuite) TestDoDispatchMessagesUnsequenced(c *C) { }, nil }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) s.settle(c) @@ -904,7 +906,7 @@ func (s *deviceMgmtMgrSuite) TestDoDispatchMessagesEvictedSequenceRejected(c *C) return &store.MessageExchangeResponse{Messages: messages}, nil }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) s.settle(c) @@ -951,7 +953,7 @@ func (s *deviceMgmtMgrSuite) TestDoDispatchMessagesBlockedSequenceRejected(c *C) return &store.MessageExchangeResponse{Messages: messages}, nil }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) s.settle(c) @@ -979,7 +981,7 @@ func (s *deviceMgmtMgrSuite) TestDoDispatchMessagesIdempotent(c *C) { return &store.MessageExchangeResponse{}, nil }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) ms := &devicemgmtstate.DeviceMgmtState{ Sequences: map[string]*devicemgmtstate.SequenceState{ @@ -1126,7 +1128,7 @@ func (s *deviceMgmtMgrSuite) TestDoApplyMessageOK(c *C) { }, nil }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) s.settle(c) @@ -1178,7 +1180,7 @@ func (s *deviceMgmtMgrSuite) TestDoApplyMessageSkipIfAlreadyFailed(c *C) { }, }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) s.settle(c) @@ -1202,7 +1204,7 @@ func (s *deviceMgmtMgrSuite) TestDoApplyMessageNoHandlerForMessageKind(c *C) { }, nil }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) s.settle(c) @@ -1233,7 +1235,7 @@ func (s *deviceMgmtMgrSuite) TestDoApplyMessageApplyError(c *C) { }, }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) s.settle(c) @@ -1264,7 +1266,7 @@ func (s *deviceMgmtMgrSuite) TestDoApplyMessageIdempotent(c *C) { }, }) - s.runner.AddHandler("queue-mgmt-response", func(*state.Task, *tomb.Tomb) error { return nil }, nil) + s.runner.AddHandler("queue-mgmt-response", noopTask, nil) ms := &devicemgmtstate.DeviceMgmtState{ Sequences: map[string]*devicemgmtstate.SequenceState{ From 31da6b27036e38f94f4b611075b030211040fc0b Mon Sep 17 00:00:00 2001 From: Stephen Mwangi Date: Wed, 1 Jul 2026 16:31:22 +0300 Subject: [PATCH 6/7] o/devicemgmtstate: improve error messages --- overlord/devicemgmtstate/devicemgmtmgr.go | 2 +- overlord/devicemgmtstate/devicemgmtmgr_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/overlord/devicemgmtstate/devicemgmtmgr.go b/overlord/devicemgmtstate/devicemgmtmgr.go index 508a921b87c..70ba66f7873 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr.go +++ b/overlord/devicemgmtstate/devicemgmtmgr.go @@ -687,7 +687,7 @@ func (m *DeviceMgmtManager) setMessageResponseFromChange(msg *RequestMessage) er body, err := handler.ResultFromChange(change) if err != nil { msg.ResponseStatus = asserts.MessageStatusError - msg.ResponseBody = map[string]any{"message": err.Error()} + msg.ResponseBody = map[string]any{"message": fmt.Sprintf("cannot process message: %v", err)} } else { msg.ResponseStatus = asserts.MessageStatusSuccess msg.ResponseBody = body diff --git a/overlord/devicemgmtstate/devicemgmtmgr_test.go b/overlord/devicemgmtstate/devicemgmtmgr_test.go index 0b22ff053d3..1a57330adc8 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr_test.go +++ b/overlord/devicemgmtstate/devicemgmtmgr_test.go @@ -1623,7 +1623,7 @@ func (s *deviceMgmtMgrSuite) TestDoQueueResponseResultFromChangeError(c *C) { sign: func(accountID, messageID string, status asserts.MessageStatus, body []byte) (*asserts.ResponseMessage, error) { c.Check(messageID, Equals, "mesg-1") c.Check(status, Equals, asserts.MessageStatusError) - c.Check(string(body), Equals, `{"message":"operation failed"}`) + c.Check(string(body), Equals, `{"message":"cannot process message: operation failed"}`) return assertstest.FakeAssertionWithBody(body, map[string]any{ "type": "response-message", From 9912fc6bd43c8b8966a9eefda85a2471be2db4be Mon Sep 17 00:00:00 2001 From: Stephen Mwangi Date: Mon, 13 Jul 2026 12:44:39 +0300 Subject: [PATCH 7/7] o/devicemgmtstate: fixes after review --- overlord/devicemgmtstate/devicemgmtmgr.go | 10 +++-- .../devicemgmtstate/devicemgmtmgr_test.go | 40 +++++++++++++++++-- 2 files changed, 44 insertions(+), 6 deletions(-) diff --git a/overlord/devicemgmtstate/devicemgmtmgr.go b/overlord/devicemgmtstate/devicemgmtmgr.go index 70ba66f7873..b75198b54ae 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr.go +++ b/overlord/devicemgmtstate/devicemgmtmgr.go @@ -636,6 +636,11 @@ func (m *DeviceMgmtManager) doQueueResponse(t *state.Task, _ *tomb.Tomb) error { return fmt.Errorf("cannot marshal response body: %w", err) } + // TODO: determine reasonable behavior for internal errors (e.g., signing or marshal failures). + // Since tasks are idempotent, a failed message will not be re-dispatched or re-applied on the + // next change, but the request message remains in state until doQueueResponse completes, + // so such failures leave it hanging indefinitely. + resAs, err := m.signer.SignResponseMessage(msg.AccountID, msg.ID(), msg.ResponseStatus, bodyBytes) if err != nil { return fmt.Errorf("cannot sign response message: %w", err) @@ -651,8 +656,7 @@ func (m *DeviceMgmtManager) doQueueResponse(t *state.Task, _ *tomb.Tomb) error { // 2. If it errors elsewhere (in validate, apply, or queue-response), we end // up not advancing Applied, which means we accumulate messages until we // hit the sequence cap. - // Refactor sequence rejection to always evict immediately and determine the - // right behavior for internal errors. + // Refactor sequence rejection to always evict immediately. if msg.SeqNum > 0 && msg.ResponseStatus == asserts.MessageStatusSuccess { ms.Sequences[msg.BaseID].Applied = msg.SeqNum } @@ -678,7 +682,7 @@ func (m *DeviceMgmtManager) setMessageResponseFromChange(msg *RequestMessage) er change := m.state.Change(msg.ApplyChangeID) if change == nil { - return fmt.Errorf("cannot find subsystem change %q", msg.ApplyChangeID) + return fmt.Errorf("internal error: cannot find subsystem change %q", msg.ApplyChangeID) } if !change.Status().Ready() { return &state.Retry{After: awaitSubsystemRetryInterval} diff --git a/overlord/devicemgmtstate/devicemgmtmgr_test.go b/overlord/devicemgmtstate/devicemgmtmgr_test.go index 1a57330adc8..2512d35c28c 100644 --- a/overlord/devicemgmtstate/devicemgmtmgr_test.go +++ b/overlord/devicemgmtstate/devicemgmtmgr_test.go @@ -1689,7 +1689,7 @@ func (s *deviceMgmtMgrSuite) TestDoQueueResponseSubsystemChangeNotFound(c *C) { s.st.Unlock() err := s.mgr.DoQueueResponse(t, &tomb.Tomb{}) s.st.Lock() - c.Assert(err, ErrorMatches, `cannot find subsystem change "16384"`) + c.Assert(err, ErrorMatches, `internal error: cannot find subsystem change "16384"`) } func (s *deviceMgmtMgrSuite) TestDoQueueResponseNoHandlerForMessageKind(c *C) { @@ -1782,11 +1782,14 @@ func (s *deviceMgmtMgrSuite) TestDoQueueResponseSubsystemChangeNotReady(c *C) { } s.mgr.SetState(ms) + changeReady := false s.mgr.RegisterHandler("test-kind", &mockMessageHandler{ resultFromChange: func(*state.Change) (map[string]any, error) { - c.Error("resultFromChange must not be called while subsystem change is not ready") + if !changeReady { + c.Error("resultFromChange must not be called while subsystem change is not ready") + } - return nil, nil + return map[string]any{"values": "ok"}, nil }, }) @@ -1799,6 +1802,37 @@ func (s *deviceMgmtMgrSuite) TestDoQueueResponseSubsystemChangeNotReady(c *C) { err := s.mgr.DoQueueResponse(t, &tomb.Tomb{}) s.st.Lock() c.Assert(err, FitsTypeOf, &state.Retry{}) + + ms, err = s.mgr.GetState() + c.Assert(err, IsNil) + c.Check(ms.ReadyResponses, HasLen, 0) + + // Now mark the subsystem change as done and verify the handler proceeds. + changeReady = true + subsysChg.SetStatus(state.DoneStatus) + + s.mgr.MockSigner(&mockSigner{ + sign: func(accountID, messageID string, status asserts.MessageStatus, body []byte) (*asserts.ResponseMessage, error) { + return assertstest.FakeAssertionWithBody(body, map[string]any{ + "type": "response-message", + "account-id": accountID, + "message-id": messageID, + "device": "serial-1.my-model.my-brand", + "status": string(status), + "body-length": strconv.Itoa(len(body)), + }).(*asserts.ResponseMessage), nil + }, + }) + + s.st.Unlock() + err = s.mgr.DoQueueResponse(t, &tomb.Tomb{}) + s.st.Lock() + c.Assert(err, IsNil) + + ms, err = s.mgr.GetState() + c.Assert(err, IsNil) + c.Check(ms.ReadyResponses, HasLen, 1) + c.Check(ms.ReadyResponses["msg1"].Format, Equals, "assertion") } func (s *deviceMgmtMgrSuite) TestDoQueueResponseSigningError(c *C) {