Handle single-name bulk slice and write cursor without clobbering metadata

api.npmjs.org returns {downloads: N, package: name} for a single-name
request rather than {name: {downloads: N}}, so a bulk slice of length 1
was silently dropped by the hash-of-hashes parser. Route single-name
slices through the per-package path instead.

The cursor write was update_column(:metadata, merged_hash), which
overwrites the whole json value with whatever this process loaded plus
the cursor key. metadata also holds rate_limit, which is tuned live via
console, so a rate_limit change made while a batch was in flight would
be lost when the cursor landed. Registry#merge_metadata_key does the
merge in SQL via jsonb || so only the one key is written.

Also add desc lines to the two rake tasks.

Author: andrew

app/models/ecosystem/npm.rb

@@ -151,6 +151,10 @@ def fetch_download_counts(names)
       scoped, unscoped = names.partition { |n| n.start_with?('@') }
       counts = {}
       unscoped.each_slice(BULK_DOWNLOADS_LIMIT) do |batch|
+        if batch.length == 1
+          scoped << batch.first
+          next
+        end
         json = get_json("https://api.npmjs.org/downloads/point/last-month/#{batch.join(',')}")
         next unless json.is_a?(Hash)
         json.each { |name, data| counts[name] = data['downloads'] if data.is_a?(Hash) }

app/models/registry.rb

@@ -87,6 +87,14 @@ def self.update_extra_counts
     all.each(&:update_extra_counts)
   end
 
+  def merge_metadata_key(key, value)
+    Registry.where(id: id).update_all([
+      "metadata = (COALESCE(metadata, '{}')::jsonb || jsonb_build_object(?, ?::text))::json",
+      key.to_s, value.to_s
+    ])
+    reload
+  end
+
   def update_download_counts(limit: 1000, top: false)
     return 0 unless ecosystem_instance.respond_to?(:fetch_download_counts)
     scope = packages.where("status IS NULL OR status <> ?", 'removed')
@@ -96,7 +104,7 @@ def update_download_counts(limit: 1000, top: false)
       cursor = (metadata || {})['download_counts_cursor'].to_s
       batch = scope.where('name > ?', cursor).order(:name).limit(limit).pluck(:id, :name)
       if batch.empty?
-        update_column(:metadata, (metadata || {}).merge('download_counts_cursor' => '')) unless cursor.blank?
+        merge_metadata_key('download_counts_cursor', '') unless cursor.blank?
         return 0
       end
     end
@@ -111,7 +119,7 @@ def update_download_counts(limit: 1000, top: false)
         Package.where(id: ids).update_all(downloads_updated_at: now)
       end
     end
-    update_column(:metadata, (metadata || {}).merge('download_counts_cursor' => batch.last.last)) unless top
+    merge_metadata_key('download_counts_cursor', batch.last.last) unless top
     batch.length
   end
 

lib/tasks/packages.rake

@@ -132,6 +132,7 @@ namespace :packages do
     end
   end
 
+  desc 'walk npm packages by name cursor and refresh download counts'
   task update_download_counts: :environment do
     with_rake_lock('packages:update_download_counts') do
       Registry.where(ecosystem: 'npm').find_each do |r|
@@ -140,6 +141,7 @@ namespace :packages do
     end
   end
 
+  desc 'refresh download counts for the top-download npm packages'
   task update_top_download_counts: :environment do
     with_rake_lock('packages:update_top_download_counts') do
       Registry.where(ecosystem: 'npm').find_each do |r|

test/models/ecosystem/npm_test.rb

@@ -160,12 +160,13 @@ class NpmTest < ActiveSupport::TestCase
   end
 
   test 'fetch_download_counts fetches scoped names individually' do
-    stub_request(:get, "https://api.npmjs.org/downloads/point/last-month/lodash")
-      .to_return(status: 200, body: '{"lodash":{"downloads":5}}', headers: { 'Content-Type' => 'application/json' })
+    stub_request(:get, "https://api.npmjs.org/downloads/point/last-month/lodash,express")
+      .to_return(status: 200, body: '{"lodash":{"downloads":5},"express":{"downloads":6}}', headers: { 'Content-Type' => 'application/json' })
     stub_request(:get, "https://api.npmjs.org/downloads/point/last-month/@scope/pkg")
       .to_return(status: 200, body: '{"downloads":7,"package":"@scope/pkg"}', headers: { 'Content-Type' => 'application/json' })
-    counts = @ecosystem.fetch_download_counts(['@scope/pkg', 'lodash'])
+    counts = @ecosystem.fetch_download_counts(['@scope/pkg', 'lodash', 'express'])
     assert_equal 5, counts['lodash']
+    assert_equal 6, counts['express']
     assert_equal 7, counts['@scope/pkg']
   end
 
@@ -179,6 +180,21 @@ class NpmTest < ActiveSupport::TestCase
     assert_equal 1, counts['pkg129']
   end
 
+  test 'fetch_download_counts routes a lone unscoped name through the single-package path' do
+    stub_request(:get, "https://api.npmjs.org/downloads/point/last-month/lodash")
+      .to_return(status: 200, body: '{"downloads":9,"package":"lodash"}', headers: { 'Content-Type' => 'application/json' })
+    assert_equal({ 'lodash' => 9 }, @ecosystem.fetch_download_counts(['lodash']))
+  end
+
+  test 'fetch_download_counts routes a single-name trailing slice through the single-package path' do
+    names = (1..129).map { |i| "pkg#{i}" }
+    stub_request(:get, %r{https://api\.npmjs\.org/downloads/point/last-month/pkg1,.*,pkg128$})
+      .to_return(status: 200, body: '{}', headers: { 'Content-Type' => 'application/json' })
+    stub_request(:get, "https://api.npmjs.org/downloads/point/last-month/pkg129")
+      .to_return(status: 200, body: '{"downloads":3,"package":"pkg129"}', headers: { 'Content-Type' => 'application/json' })
+    assert_equal 3, @ecosystem.fetch_download_counts(names)['pkg129']
+  end
+
   test 'fetch_download_counts ignores non-hash bulk response' do
     stub_request(:get, "https://api.npmjs.org/downloads/point/last-month/a,b")
       .to_return(status: 500, body: 'null', headers: { 'Content-Type' => 'application/json' })

test/models/registry_test.rb

@@ -427,6 +427,22 @@ class RegistryTest < ActiveSupport::TestCase
     assert_equal 3, stat_2022.packages_count
   end
 
+  test 'merge_metadata_key preserves keys written by another connection after load' do
+    @registry.update!(metadata: { 'rate_limit' => 1 })
+    stale = Registry.find(@registry.id)
+    Registry.where(id: @registry.id).update_all("metadata = (metadata::jsonb || '{\"rate_limit\":5}'::jsonb)::json")
+    stale.merge_metadata_key('download_counts_cursor', 'zzz')
+    fresh = Registry.find(@registry.id)
+    assert_equal 5, fresh.metadata['rate_limit']
+    assert_equal 'zzz', fresh.metadata['download_counts_cursor']
+  end
+
+  test 'merge_metadata_key handles empty metadata' do
+    @registry.update_column(:metadata, nil)
+    @registry.merge_metadata_key('download_counts_cursor', 'abc')
+    assert_equal 'abc', @registry.metadata['download_counts_cursor']
+  end
+
   test 'throttled_ids returns registries with a rate_limit' do
     Registry.reset_throttle_cache
     throttled = Registry.create!(name: 'npmjs.org', url: 'https://registry.npmjs.org', ecosystem: 'npm', metadata: { 'rate_limit' => 1 })

test/sidekiq/update_download_counts_worker_test.rb

@@ -53,6 +53,12 @@ class UpdateDownloadCountsWorkerTest < ActiveSupport::TestCase
     assert_nil @registry.reload.metadata['download_counts_cursor']
   end
 
+  test "cursor write goes through merge_metadata_key" do
+    @registry.expects(:merge_metadata_key).with('download_counts_cursor', 'bbb')
+    @registry.ecosystem_instance.expects(:fetch_download_counts).with(['aaa', 'bbb']).returns({})
+    @registry.update_download_counts(limit: 10)
+  end
+
   test "update_download_counts is a no-op when ecosystem has no fetch_download_counts" do
     r = Registry.create!(name: 'crates.io', url: 'https://crates.io', ecosystem: 'cargo')
     r.packages.create!(name: 'x', ecosystem: 'cargo')