fix(telemetry): handle LATEST across streaming, limits, writes and history

openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb

@@ -47,13 +47,23 @@ def build_item_collection(collection, data, start_time: nil, end_time: nil, scop
       # Unwrap LATEST packet into individual items
       parsed = StreamingKey.parse(key, item_key: true)
       if parsed.packet_name == 'LATEST'
+        # The item to packet map only covers telemetry
+        raise "LATEST is only supported for telemetry: #{key}" if parsed.cmd_or_tlm == :CMD
         item_map = OpenC3::TargetModel.get_item_to_packet_map(parsed.target_name, scope: scope)
         packet_names = item_map[parsed.item_name]
         raise "Item '#{parsed.target_name} LATEST #{parsed.item_name}' does not exist for scope: #{scope}" unless packet_names
+        # Skip packets the user can't access unless they can't access any of them
+        auth_error = nil
+        added = false
         packet_names.each do |packet_name|
           new_key = parsed.with(packet_name: packet_name).to_key_string
-          collection.add(StreamingObject.new(new_key, start_time, end_time, item_key: item_key, scope: scope, token: token))
+          collection.add(StreamingObject.new(new_key, start_time, end_time, item_key: item_key, source_key: key, scope: scope, token: token))
+          added = true
+        rescue OpenC3::AuthError, OpenC3::ForbiddenError => e
+          OpenC3::Logger.info("Skipping unauthorized LATEST packet: #{new_key}")
+          auth_error = e
         end
+        raise auth_error if auth_error and not added
       else
         collection.add(StreamingObject.new(key, start_time, end_time, item_key: item_key, scope: scope, token: token))
       end

openc3-cosmos-cmd-tlm-api/app/models/streaming_object.rb

@@ -40,11 +40,14 @@ class StreamingObject
   attr_reader :id
   attr_reader :realtime
   attr_reader :item_key
+  # The key as subscribed, which differs from key when a LATEST item expands to a concrete packet
+  attr_reader :source_key
 
-  def initialize(key, start_time_nsec, end_time_nsec, item_key: nil, scope:, token: nil)
+  def initialize(key, start_time_nsec, end_time_nsec, item_key: nil, source_key: nil, scope:, token: nil)
     key = key.upcase
     @key = key
     @item_key = item_key
+    @source_key = source_key ? source_key.upcase : key
     parsed = StreamingKey.parse(key, item_key: !!item_key)
     @stream_mode = parsed.stream_mode
     @cmd_or_tlm = parsed.cmd_or_tlm
@@ -76,7 +79,9 @@ def initialize(key, start_time_nsec, end_time_nsec, item_key: nil, scope:, token
     @offset = "0-0"
     @offset = OpenC3::Topic.get_last_offset(@topic, db_shard: @db_shard) unless @start_time
     if @item_key
-      @id = 'ITEM__' + key
+      # Include the source_key and item_key so a LATEST item (which expands to concrete
+      # packet keys) doesn't collide with an explicit subscription to the same packet item
+      @id = "ITEM__#{key}__#{@source_key}__#{@item_key}"
     else
       @id = 'PACKET__' + key
     end

openc3-cosmos-cmd-tlm-api/app/models/streaming_object_collection.rb

@@ -58,36 +58,54 @@ def add(object)
   def remove(object)
     @mutex.synchronize do
       found_object = @objects_by_id[object.id]
-      if found_object
-        @objects.delete(found_object)
-        @objects_by_id.delete(found_object.id)
-        item_objects = @item_objects_by_topic[object.topic]
-        item_objects.delete(found_object) if item_objects
-        packet_objects = @packet_objects_by_topic[object.topic]
-        packet_objects.delete(found_object) if packet_objects
-        if item_objects
-          if item_objects.length == 0
-            if packet_objects
-              if packet_objects.length == 0
-                # Nothing left in either for this topic
-                @topics_and_offsets.delete(object.topic)
-              end
-            else
-              # Just item_objects and nothing left
-              @topics_and_offsets.delete(object.topic)
-            end
-          end
-        else
-          if packet_objects
-            if packet_objects.length == 0
-              # Just packet_objects and nothing left
-              @topics_and_offsets.delete(object.topic)
-            end
-          else
-            # Neither objects - Shouldn't happen
+      # Items can be added as [key, item_key] pairs but removed by key alone
+      # (the item_key defaults to the key) so remove every item object subscribed with the same key.
+      # Match on source_key so removing an explicit item doesn't also remove a
+      # LATEST subscription which expanded to the same concrete key (and vice versa)
+      key_only = (object.item_key and object.item_key.upcase == object.source_key)
+      if object.item_key and (key_only or not found_object)
+        # An explicit item_key that wasn't found exactly must also match on item_key
+        # so an unknown [key, item_key] pair can't remove other subscriptions
+        @objects.select do |obj|
+          obj.item_key and obj.source_key == object.source_key and (key_only or obj.item_key == object.item_key)
+        end.each do |obj|
+          _remove_found(obj)
+        end
+      elsif found_object
+        _remove_found(found_object)
+      end
+    end
+  end
+
+  # Must be called with @mutex held
+  def _remove_found(object)
+    @objects.delete(object)
+    @objects_by_id.delete(object.id)
+    item_objects = @item_objects_by_topic[object.topic]
+    item_objects.delete(object) if item_objects
+    packet_objects = @packet_objects_by_topic[object.topic]
+    packet_objects.delete(object) if packet_objects
+    if item_objects
+      if item_objects.length == 0
+        if packet_objects
+          if packet_objects.length == 0
+            # Nothing left in either for this topic
             @topics_and_offsets.delete(object.topic)
           end
+        else
+          # Just item_objects and nothing left
+          @topics_and_offsets.delete(object.topic)
+        end
+      end
+    else
+      if packet_objects
+        if packet_objects.length == 0
+          # Just packet_objects and nothing left
+          @topics_and_offsets.delete(object.topic)
         end
+      else
+        # Neither objects - Shouldn't happen
+        @topics_and_offsets.delete(object.topic)
       end
     end
   end

openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb

@@ -551,6 +551,124 @@ def start_fake_thread(stop_delay)
     end
   end
 
+  describe '#build_item_collection' do
+    before(:each) do
+      allow(OpenC3::TargetModel).to receive(:get_item_to_packet_map).with('INST', scope: 'DEFAULT').and_return({
+        'TEMP1' => ['HEALTH_STATUS', 'PARAMS'],
+      })
+    end
+
+    let(:latest_key) { 'DECOM__TLM__INST__LATEST__TEMP1__CONVERTED' }
+    let(:explicit_key) { 'DECOM__TLM__INST__HEALTH_STATUS__TEMP1__CONVERTED' }
+
+    def build(items)
+      collection = StreamingObjectCollection.new
+      @api.build_item_collection(collection, { 'items' => items }, scope: 'DEFAULT')
+      collection
+    end
+
+    it 'expands LATEST into one object per packet keyed by the LATEST key' do
+      collection = build([latest_key])
+      expect(collection.objects.map { |o| [o.key, o.item_key] }).to eq([
+        ['DECOM__TLM__INST__HEALTH_STATUS__TEMP1__CONVERTED', latest_key],
+        ['DECOM__TLM__INST__PARAMS__TEMP1__CONVERTED', latest_key],
+      ])
+    end
+
+    it 'keeps a LATEST item and an explicit item for the same packet' do
+      collection = build([latest_key, explicit_key])
+      expect(collection.length).to eq(3)
+      expect(collection.objects.select { |o| o.key == explicit_key }.map(&:item_key)).to contain_exactly(latest_key, explicit_key)
+    end
+
+    it 'removing the LATEST item leaves the explicit item' do
+      collection = build([latest_key, explicit_key])
+      build([latest_key]).objects.each { |o| collection.remove(o) }
+      expect(collection.objects.map { |o| [o.key, o.item_key] }).to eq([[explicit_key, explicit_key]])
+    end
+
+    it 'removing the explicit item leaves the LATEST item' do
+      collection = build([latest_key, explicit_key])
+      build([explicit_key]).objects.each { |o| collection.remove(o) }
+      expect(collection.objects.map(&:item_key).uniq).to eq([latest_key])
+      expect(collection.length).to eq(2)
+    end
+
+    it 'rejects LATEST for commands' do
+      expect { build(['DECOM__CMD__INST__LATEST__TEMP1__CONVERTED']) }.to raise_error(/LATEST is only supported for telemetry/)
+    end
+
+    it 'skips LATEST packets the user cannot access' do
+      allow_any_instance_of(StreamingObject).to receive(:authorize) do |_object, packet_name:, **|
+        raise OpenC3::ForbiddenError, 'forbidden' if packet_name == 'PARAMS'
+      end
+      expect(build([latest_key]).objects.map(&:packet_name)).to eq(['HEALTH_STATUS'])
+    end
+
+    it 'raises when the user cannot access any LATEST packet' do
+      allow_any_instance_of(StreamingObject).to receive(:authorize).and_raise(OpenC3::ForbiddenError, 'forbidden')
+      expect { build([latest_key]) }.to raise_error(OpenC3::ForbiddenError)
+    end
+
+    it 'removes an item added with an item_key when removed by key alone' do
+      collection = build([[explicit_key, '0']])
+      build([explicit_key]).objects.each { |o| collection.remove(o) }
+      expect(collection).to be_empty
+    end
+
+    it 'removing an explicit item by key alone leaves the LATEST item' do
+      collection = build([latest_key, [explicit_key, '0']])
+      build([explicit_key]).objects.each { |o| collection.remove(o) }
+      expect(collection.objects.map(&:item_key).uniq).to eq([latest_key])
+      expect(collection.length).to eq(2)
+    end
+
+    it 'removing a LATEST item by key alone leaves the explicit item' do
+      collection = build([[latest_key, '0'], explicit_key])
+      build([latest_key]).objects.each { |o| collection.remove(o) }
+      expect(collection.objects.map { |o| [o.key, o.item_key] }).to eq([[explicit_key, explicit_key]])
+    end
+
+    it 'removes every item_key for a key when removed by key alone' do
+      collection = build([explicit_key, [explicit_key, '0']])
+      expect(collection.length).to eq(2)
+      build([explicit_key]).objects.each { |o| collection.remove(o) }
+      expect(collection).to be_empty
+    end
+
+    it 'removing a key with an item_key only removes that item_key' do
+      collection = build([explicit_key, [explicit_key, '0']])
+      build([[explicit_key, '0']]).objects.each { |o| collection.remove(o) }
+      expect(collection.objects.map(&:item_key)).to eq([explicit_key])
+    end
+
+    it 'removing an unknown item_key does not remove other item_keys' do
+      collection = build([explicit_key, [explicit_key, '0']])
+      build([[explicit_key, '1']]).objects.each { |o| collection.remove(o) }
+      expect(collection.objects.map(&:item_key)).to contain_exactly(explicit_key, '0')
+    end
+
+    it 'falls back to matching source_key and item_key when the exact object is not found' do
+      allow(OpenC3::TargetModel).to receive(:get_item_to_packet_map).with('INST', scope: 'DEFAULT').and_return({
+        'TEMP1' => ['PARAMS'],
+      })
+      collection = build([[latest_key, '0'], [latest_key, '1']])
+      allow(OpenC3::TargetModel).to receive(:get_item_to_packet_map).with('INST', scope: 'DEFAULT').and_return({
+        'TEMP1' => ['HEALTH_STATUS'],
+      })
+      # The HEALTH_STATUS object was never added so the removal falls back to the sweep
+      build([[latest_key, '0']]).objects.each { |o| collection.remove(o) }
+      expect(collection.objects.map(&:item_key)).to eq(['1'])
+    end
+
+    it 'keeps a LATEST item and an explicit item with the same item_key' do
+      collection = build([[latest_key, '0'], [explicit_key, '0']])
+      expect(collection.length).to eq(3)
+      build([[latest_key, '0']]).objects.each { |o| collection.remove(o) }
+      expect(collection.objects.map { |o| [o.key, o.source_key] }).to eq([[explicit_key, explicit_key]])
+    end
+  end
+
   describe 'add with COSMOS_ALL packets and authorization' do
     it 'skips unauthorized packets when using COSMOS_ALL' do
       $openc3_authorize = true