fix(atelet): Supersede stale system-info volume registrations in atelet - #1684

+// registration is the entry Register created. Undo removes that entry and
+// nothing else, so a caller that fails after registering cannot drop a
+// newer registration for the same actor.
+type registration struct {
+     r     *systemInfoVolumeRefresher
+     actor *registeredActor
+}
+
+// Undo drops the registration unless a later Register superseded it.
+// Safe on a nil handle so callers can defer it before Register runs.
+func (h *registration) Undo() {
+     if h == nil {
+             return
+     }
+     h.r.drop(h.actor)
+     h.actor.mu.Lock()
+     h.actor.stale = true
+     h.actor.mu.Unlock()
+}
+
+// drop removes actor from the registry if it is still the live entry.
+func (r *systemInfoVolumeRefresher) drop(actor *registeredActor) {
+     r.mu.Lock()
+     if r.actors[actor.uid] == actor {
+             delete(r.actors, actor.uid)
+     }
+     r.mu.Unlock()
+}

 // Register records actorUID's system-info volumes and writes their contents
-// from current cluster state. If actorUID is already registered (for example
-// after a worker pod crash left a stale entry without Terminate), the previous
-// registration is superseded.
-func (r *systemInfoVolumeRefresher) Register(actorUID string, ref resources.ActorRef, volumes []*systemInfoVolume) error {
+// from current cluster state. A second Register for the same actorUID
+// supersedes the first, which stops receiving refreshes. A failed Register
+// leaves no entry behind.
+func (r *systemInfoVolumeRefresher) Register(actorUID string, ref resources.ActorRef, volumes []*systemInfoVolume) (*registration, error) {
      actor := &registeredActor{uid: actorUID, ref: ref, volumes: volumes}
      // Held until the initial write finishes so a refresh cannot interleave.
      actor.mu.Lock()
      defer actor.mu.Unlock()
@@
      if prev != nil {
              prev.mu.Lock()
              prev.stale = true
              prev.mu.Unlock()
      }

      for _, v := range volumes {
              if err := r.write(ref, actorUID, v); err != nil {
-                     return fmt.Errorf("while populating system-info volume %q: %w", v.Name, err)
+                     // actor.mu is already held here, so this cannot go through Undo.
+                     r.drop(actor)
+                     actor.stale = true
+                     return nil, fmt.Errorf("while populating system-info volume %q: %w", v.Name, err)
              }
      }
-     return nil
+     return &registration{r: r, actor: actor}, nil

--- a/cmd/atelet/main.go
+++ b/cmd/atelet/main.go
@@ func (s *AteomHerder) Run(
-     defer func() {
-             if err != nil {
-                     s.systemInfoVolumes.Deregister(actorUID)
-             }
-     }()
-     if err := s.systemInfoVolumes.Register(actorUID, actorRef, systemInfoVolumesFor(actorUID, req.GetSpec())); err != nil {
+     reg, err := s.systemInfoVolumes.Register(actorUID, actorRef, systemInfoVolumesFor(actorUID, req.GetSpec()))
+     if err != nil {
              return nil, err
      }
+     defer func() {
+             if err != nil {
+                     reg.Undo()
+             }
+     }()
      if err := s.prepareOCIBundles(ctx, actorUID, actorRef,
@@ func (s *AteomHerder) Restore(
-     // Undo the Register if the restore fails.
+     // Undo the Register if the restore fails. The prep leg writes reg and
+     // nothing reads it before g.Wait returns.
+     var reg *registration
      defer func() {
              if err != nil {
-                     s.systemInfoVolumes.Deregister(actorUID)
+                     reg.Undo()
              }
      }()
@@
-             if err = s.systemInfoVolumes.Register(actorUID, actorRef, systemInfoVolumesFor(actorUID, req.GetSpec())); err != nil {
+             if reg, err = s.systemInfoVolumes.Register(actorUID, actorRef, systemInfoVolumesFor(actorUID, req.GetSpec())); err != nil {