Class: Webhookdb::SyncTarget::HttpRoutine

Inherits:

Routine

• Object
• Routine
• Webhookdb::SyncTarget::HttpRoutine

Defined in:

lib/webhookdb/sync_target.rb

Instance Attribute Summary

Attributes inherited from Routine

#last_synced_at, #now, #replicator, #sync_target, #timestamp_expr

Instance Method Summary

• #_flush_http_chunk(chunk) ⇒ Object
• #initialize ⇒ HttpRoutine constructor

  A new instance of HttpRoutine.

• #run ⇒ Object

Methods inherited from Routine

#dataset_to_sync, #perform_db_op, #record, #to_ms, #with_stat

Constructor Details

#initialize ⇒ HttpRoutine

Returns a new instance of HttpRoutine.

# File 'lib/webhookdb/sync_target.rb', line 455

def initialize(*)
  super
  @inflight_timestamps = []
  @cleanurl, @authparams = Webhookdb::Http.extract_url_auth(self.sync_target.connection_url)
  @threadpool = if self.sync_target.parallelism.zero?
                  Webhookdb::Concurrent::SerialPool.new
    else
      Webhookdb::Concurrent::ParallelizedPool.new(self.sync_target.parallelism)
  end
  @mutex = Thread::Mutex.new
end

Instance Method Details

#_flush_http_chunk(chunk) ⇒ Object

# File 'lib/webhookdb/sync_target.rb', line 523

def _flush_http_chunk(chunk)
  chunk_ts = chunk.last.fetch(self.replicator.timestamp_column.name)
  @mutex.synchronize do
    @inflight_timestamps << chunk_ts
    @inflight_timestamps.sort!
  end
  sint = self.sync_target.service_integration
  body = {
    rows: chunk,
    integration_id: sint.opaque_id,
    integration_service: sint.service_name,
    table: sint.table_name,
    sync_timestamp: self.now,
  }
  @threadpool.post do
    self.with_stat do
      Webhookdb::Http.post(
        @cleanurl,
        body,
        timeout: sint.organization.sync_target_timeout,
        logger: self.sync_target.logger,
        basic_auth: @authparams,
      )
    end
    # On success, we want to commit the latest timestamp we sent to the client,
    # so it can be recorded. Then in the case of an error on later rows,
    # we won't re-sync rows we've already processed (with earlier updated timestamps).
    @mutex.synchronize do
      this_ts_idx = @inflight_timestamps.index { |t| t == chunk_ts }
      raise Webhookdb::InvariantViolation, "timestamp no longer found!?" if this_ts_idx.nil?
      # However, we only want to record the timestamp if this request is the earliest inflight request;
      # ie, if a later request finishes before an earlier one, we don't want to record the timestamp
      # of the later request as 'finished' since the earlier one didn't finish.
      # This does mean though that, if the earliest request errors, we'll throw away the work
      # done by the later request.
      # Note that each row can only appear in a sync once, even if it is modified after the sync starts;
      # thus, parallel httpsync should be fine for most clients to handle,
      # since race conditions *on the same row* cannot happen even with parallel httpsync.
      self.record(chunk_ts) if this_ts_idx.zero?
      @inflight_timestamps.delete_at(this_ts_idx)
    end
  end
end

#run ⇒ Object

# File 'lib/webhookdb/sync_target.rb', line 467

def run
  timeout_at = Time.now + Webhookdb::SyncTarget.max_transaction_seconds
  page_size = self.sync_target.page_size
  sync_result = :complete
  self.dataset_to_sync do |ds|
    chunk = []
    ds.paged_each(rows_per_fetch: page_size, cursor_name: "synctarget_#{self.sync_target.id}_cursor") do |row|
      chunk << row
      if chunk.size >= page_size
        # Do not share chunks across threads
        self._flush_http_chunk(chunk.dup)
        chunk.clear
        if Time.now >= timeout_at && Thread.current[:sidekiq_context]
          # If we've hit the timeout, stop any further syncing
          sync_result = :timeout
          break
        end
      end
    end
    self._flush_http_chunk(chunk) unless chunk.empty?
    @threadpool.join
    case sync_result
      when :timeout
        # If the sync timed out, use the last recorded sync timestamp,
        # and re-enqueue the job, so the sync will pick up where it left off.
        self.sync_target.logger.info("sync_target_transaction_timeout", self.sync_target.log_tags)
        Webhookdb::Jobs::SyncTargetRunSync.perform_async(self.sync_target.id)
      else
        # The sync completed normally.
        # Save 'now' as the timestamp, rather than the last updated row.
        # This is important because other we'd keep trying to sync the last row synced.
        self.record(self.now)
    end
  end
rescue Webhookdb::Concurrent::Timeout => e
  # This should never really happen, but it does, so record it while we debug it.
  self.perform_db_op do
    self.sync_target.save_changes
  end
  self.sync_target.logger.error("sync_target_pool_timeout_error", self.sync_target.log_tags, e)
rescue StandardError => e
  # Errors talking to the http server are handled well so no need to re-raise.
  # We already committed the last page that was successful,
  # so we can just stop syncing at this point to try again later.
  raise e unless e.is_a?(Webhookdb::Http::Error) || Webhookdb::SyncTarget.transport_error?(e)
  self.perform_db_op do
    # Save any outstanding stats.
    self.sync_target.save_changes
  end
  # Don't spam our logs with downstream errors
  idem_key = "sync_target_http_error-#{self.sync_target.id}-#{e.class.name}"
  Webhookdb::Idempotency.every(1.hour).in_memory.under_key(idem_key) do
    self.sync_target.logger.warn("sync_target_http_error", self.sync_target.log_tags, e)
  end
end