Fix event drop by fetching 3 pages with ETag optimization

crawler/Gemfile

@@ -1,7 +1,7 @@
 source 'http://rubygems.org'
 
-gem 'eventmachine', :git => 'git://github.com/eventmachine/eventmachine.git'
-gem 'em-http-request', :git => 'git://github.com/igrigorik/em-http-request.git'
+gem 'eventmachine', :git => 'https://github.com/eventmachine/eventmachine.git'
+gem 'em-http-request', :git => 'https://github.com/igrigorik/em-http-request.git'
 
 gem 'yajl-ruby', :require => 'yajl'
 gem 'em-stathat'

crawler/Gemfile.lock

@@ -1,11 +1,11 @@
 GIT
-  remote: git://github.com/eventmachine/eventmachine.git
+  remote: https://github.com/eventmachine/eventmachine.git
   revision: 98ff494aac2279af2675fd8dd49cc5f130f8b236
   specs:
     eventmachine (1.2.5)
 
 GIT
-  remote: git://github.com/igrigorik/em-http-request.git
+  remote: https://github.com/igrigorik/em-http-request.git
   revision: 6061430336bd1421b25c2244d5b85e4cdfcfb3f7
   specs:
     em-http-request (1.1.5)

crawler/crawler.rb

@@ -12,7 +12,7 @@
 ## Setup
 ##
 
-PAGE_LIMIT = 500
+PAGE_LIMIT = 100
 
 StatHat.config do |c|
   c.ukey  = ENV['STATHATKEY']
@@ -44,68 +44,122 @@
 
   @latest = []
   @latest_key = lambda { |e| "#{e['id']}" }
+  @etags = {}  # Track ETags for each page
 
   process = Proc.new do
-      req = HttpRequest.new("https://api.github.com/events?per_page=#{PAGE_LIMIT}", {
-        :inactivity_timeout => 5,
-        :connect_timeout => 5
-      }).get({
+    # First, probe page 1 with conditional GET
+    req1 = HttpRequest.new("https://api.github.com/events?per_page=#{PAGE_LIMIT}&page=1", {
+      :inactivity_timeout => 5,
+      :connect_timeout => 5
+    }).get({
       :head => {
         'user-agent' => 'gharchive.org',
-        'Authorization' => 'token ' + ENV['GITHUB_TOKEN']
-      }
+        'Authorization' => 'token ' + ENV['GITHUB_TOKEN'],
+        'If-None-Match' => @etags[1]
+      }.compact
     })
 
-    req.callback do
+    req1.callback do
       begin
-        latest = Yajl::Parser.parse(req.response)
-        urls = latest.collect(&@latest_key)
-        new_events = latest.reject {|e| @latest.include? @latest_key.call(e)}
-
-        @latest = urls
-
-        # Determine archive filename based on current time, before processing events
-        current_processing_time = Time.now
-        timestamp = current_processing_time.strftime('%Y-%m-%d-%-k')
-        archive = "data/#{timestamp}.json"
-
-        # Open or rotate file based on the current time's archive path
-        if @file.nil? || (archive != @file.to_path)
-          if !@file.nil?
-            @log.info "Rotating archive. Current: #{@file.to_path}, New: #{archive}"
-            @file.close
-          end
-          @file = File.new(archive, "a+")
+        # If page 1 hasn't changed (304 Not Modified), skip this cycle
+        if req1.response_header.status == 304
+          @log.debug "Page 1 not modified, skipping"
+          EM.add_timer(0.2, &process)
+          return
         end
 
-        new_events.each do |event|
-          @file.puts(Yajl::Encoder.encode(Obfuscate.email(event)))
-        end
+        # Page 1 changed, update ETag and fetch pages 2 & 3
+        @etags[1] = req1.response_header.etag
+
+        # Fetch pages 2 and 3 in parallel (GitHub only provides up to 300 events)
+        multi = EM::MultiRequest.new
+
+        req2 = HttpRequest.new("https://api.github.com/events?per_page=#{PAGE_LIMIT}&page=2", {
+          :inactivity_timeout => 5,
+          :connect_timeout => 5
+        }).get({
+          :head => {
+            'user-agent' => 'gharchive.org',
+            'Authorization' => 'token ' + ENV['GITHUB_TOKEN'],
+            'If-None-Match' => @etags[2]
+          }.compact
+        })
+
+        req3 = HttpRequest.new("https://api.github.com/events?per_page=#{PAGE_LIMIT}&page=3", {
+          :inactivity_timeout => 5,
+          :connect_timeout => 5
+        }).get({
+          :head => {
+            'user-agent' => 'gharchive.org',
+            'Authorization' => 'token ' + ENV['GITHUB_TOKEN'],
+            'If-None-Match' => @etags[3]
+          }.compact
+        })
+
+        multi.add(:page2, req2)
+        multi.add(:page3, req3)
+
+        multi.callback do
+          # Update ETags
+          @etags[2] = req2.response_header.etag if req2.response_header.status == 200
+          @etags[3] = req3.response_header.etag if req3.response_header.status == 200
+
+          # Parse all responses
+          page1_events = Yajl::Parser.parse(req1.response)
+          page2_events = req2.response_header.status == 200 ? Yajl::Parser.parse(req2.response) : []
+          page3_events = req3.response_header.status == 200 ? Yajl::Parser.parse(req3.response) : []
+
+          # Merge all events from the 3 pages (GitHub's max is 300 events)
+          latest = page1_events + page2_events + page3_events
+          urls = latest.collect(&@latest_key)
+          new_events = latest.reject {|e| @latest.include? @latest_key.call(e)}
+
+          @latest = urls
+
+          # Determine archive filename based on current time, before processing events
+          current_processing_time = Time.now
+          timestamp = current_processing_time.strftime('%Y-%m-%d-%-k')
+          archive = "data/#{timestamp}.json"
+
+          # Open or rotate file based on the current time's archive path
+          if @file.nil? || (archive != @file.to_path)
+            if !@file.nil?
+              @log.info "Rotating archive. Current: #{@file.to_path}, New: #{archive}"
+              @file.close
+            end
+            @file = File.new(archive, "a+")
+          end
 
-        remaining = req.response_header.raw['X-RateLimit-Remaining']
-        reset = Time.at(req.response_header.raw['X-RateLimit-Reset'].to_i)
-        @log.info "Found #{new_events.size} new events: #{new_events.collect(&@latest_key)}, API: #{remaining}, reset: #{reset}"
+          new_events.each do |event|
+            @file.puts(Yajl::Encoder.encode(Obfuscate.email(event)))
+          end
 
-        if new_events.size >= PAGE_LIMIT
-          @log.info "Missed records.."
-        end
+          remaining = req1.response_header.raw['X-RateLimit-Remaining']
+          reset = Time.at(req1.response_header.raw['X-RateLimit-Reset'].to_i)
+          @log.info "Found #{new_events.size} new events (page1: #{page1_events.size}, page2: #{page2_events.size}, page3: #{page3_events.size}), API: #{remaining}, reset: #{reset}"
 
-        StatHat.new.ez_count('Github Events', new_events.size)
+          if new_events.size >= (PAGE_LIMIT * 3)
+            @log.warn "Potentially missed records - got #{new_events.size} new events (at GitHub's 300 event limit)"
+          end
+
+          StatHat.new.ez_count('Github Events', new_events.size)
+
+          EM.add_timer(0.2, &process)
+        end
 
       rescue Exception => e
         @log.error "Failed to process response"
-        @log.error "Response: #{req.response}"
-        @log.error "Response headers: #{req.response_header}"
+        @log.error "Response page 1: #{req1.response}"
+        @log.error "Response headers: #{req1.response_header}"
         @log.error "Processing exception: #{e}, #{e.backtrace.first(5)}"
-      ensure
         EM.add_timer(0.75, &process)
       end
     end
 
-    req.errback do
-      @log.error "Error: #{req.response_header.status}, \
-                  header: #{req.response_header}, \
-                  response: #{req.response}"
+    req1.errback do
+      @log.error "Error fetching page 1: #{req1.response_header.status}, \
+                  header: #{req1.response_header}, \
+                  response: #{req1.response}"
 
       EM.add_timer(0.75, &process)
     end