overleaf/services/document-updater/test/stress/coffee/run.coffee

DocUpdaterClient = require "../../acceptance/coffee/helpers/DocUpdaterClient"
# MockTrackChangesApi = require "../../acceptance/js/helpers/MockTrackChangesApi"
# MockWebApi = require "../../acceptance/js/helpers/MockWebApi"
assert = require "assert"
async = require "async"

insert = (string, pos, content) ->
	result = string.slice(0, pos) + content + string.slice(pos)
	return result

transform = (op1, op2) ->
	if op2.p < op1.p
		return {
			p: op1.p + op2.i.length
			i: op1.i
		}
	else
		return op1

class StressTestClient
	constructor: (@options = {}) ->
		@options.updateDelay ?= 200
		@project_id = @options.project_id or DocUpdaterClient.randomId()
		@doc_id = @options.doc_id or DocUpdaterClient.randomId()
		@pos = @options.pos or 0
		@content = @options.content or ""

		@client_id = DocUpdaterClient.randomId()
		@version = @options.version or 0
		@inflight_op = null
		@charCode = 0
		
		@counts = {
			conflicts: 0
			local_updates: 0
			remote_updates: 0
			max_delay: 0
		}
		
		DocUpdaterClient.subscribeToAppliedOps (channel, update) =>
			update = JSON.parse(update)
			if update.error?
				console.error new Error("Error from server: '#{update.error}'")
				return
			if update.doc_id == @doc_id
				@processReply(update)
	
	sendUpdate: () ->
		data = String.fromCharCode(65 + @charCode++ % 26)
		@content = insert(@content, @pos, data)
		@inflight_op = {
			i: data
			p: @pos++
		}
		@resendUpdate()
		@inflight_op_sent = Date.now()
	
	resendUpdate: () ->
		assert(@inflight_op?)
		DocUpdaterClient.sendUpdate(
			@project_id, @doc_id
			{
				doc: @doc_id
				op: [@inflight_op]
				v: @version
				meta:
					source: @client_id
				dupIfSource: [@client_id]
			}
		)
		@update_timer = setTimeout () =>
			console.log "[#{new Date()}] \t[#{@client_id.slice(0,4)}] WARN: Resending update after 5 seconds"
			@resendUpdate()
		, 5000
	
	processReply: (update) ->
		if update.op.v != @version
			if update.op.v < @version
				console.log "[#{new Date()}] \t[#{@client_id.slice(0,4)}] WARN: Duplicate ack (already seen version)"
				return
			else
				console.error "[#{new Date()}] \t[#{@client_id.slice(0,4)}] ERROR: Version jumped ahead (client: #{@version}, op: #{update.op.v})"
		@version++
		if update.op.meta.source == @client_id
			if @inflight_op?
				@counts.local_updates++
				@inflight_op = null
				clearTimeout @update_timer
				delay = Date.now() - @inflight_op_sent
				@counts.max_delay = Math.max(@counts.max_delay, delay)
				@continue()
			else
				console.log "[#{new Date()}] \t[#{@client_id.slice(0,4)}] WARN: Duplicate ack"
		else
			assert(update.op.op.length == 1)
			@counts.remote_updates++
			external_op = update.op.op[0]
			if @inflight_op?
				@counts.conflicts++
				@inflight_op = transform(@inflight_op, external_op)
				external_op = transform(external_op, @inflight_op)
			if external_op.p < @pos
				@pos += external_op.i.length
			@content = insert(@content, external_op.p, external_op.i)
	
	continue: () ->
		if @updateCount > 0
			@updateCount--
			setTimeout () =>
				@sendUpdate()
			, @options.updateDelay * ( 0.5 + Math.random() )
		else
			@updateCallback()
	
	runForNUpdates: (n, callback = (error) ->) ->
		@updateCallback = callback
		@updateCount = n
		@continue()
	
	check: (callback = (error) ->) ->
		DocUpdaterClient.getDoc @project_id, @doc_id, (error, res, body) =>
			throw error if error?
			if !body.lines?
				return console.error "[#{new Date()}] \t[#{@client_id.slice(0,4)}] ERROR: Invalid response from get doc (#{doc_id})", body
			content = body.lines.join("\n")
			version = body.version
			if content != @content
				if version == @version
					console.error "[#{new Date()}] \t[#{@client_id.slice(0,4)}] Error: Client content does not match server."
					console.error "Server: #{content.split('a')}"
					console.error "Client: #{@content.split('a')}"
				else
					console.error "[#{new Date()}] \t[#{@client_id.slice(0,4)}] Error: Version mismatch (Server: '#{version}', Client: '#{@version}')"

			if !@isContentValid(@content)
				for chunk, i in @content.split("")
					if chunk? and chunk != "a"
						console.log chunk, i
				throw new Error("bad content")
			callback()

	isChunkValid: (chunk) ->
		char = 0
		for letter, i in chunk
			if letter.charCodeAt(0) != 65 + i % 26
				console.error "[#{new Date()}] \t[#{@client_id.slice(0,4)}] Invalid Chunk:", chunk
				return false
		return true

	isContentValid: (content) ->
		for chunk in content.split('a')
			if chunk? and chunk != ""
				if !@isChunkValid(chunk)
					
					console.error "[#{new Date()}] \t[#{@client_id.slice(0,4)}] Invalid content", content
					return false
		return true


checkDocument = (project_id, doc_id, clients, callback = (error) ->) ->
	jobs = clients.map (client) ->
		(cb) -> client.check cb
	async.parallel jobs, callback

printSummary = (doc_id, clients) ->
	slot = require('cluster-key-slot')
	now = new Date()
	console.log "[#{now}] [#{doc_id.slice(0,4)} (slot: #{slot(doc_id)})] #{clients.length} clients..."
	for client in clients
		console.log "[#{now}] \t[#{client.client_id.slice(0,4)}] { local: #{client.counts.local_updates }, remote: #{client.counts.remote_updates}, conflicts: #{client.counts.conflicts}, max_delay: #{client.counts.max_delay} }"
		client.counts = {
			local_updates: 0
			remote_updates: 0
			conflicts: 0
			max_delay: 0
		}

CLIENT_COUNT = parseInt(process.argv[2], 10)
UPDATE_DELAY = parseInt(process.argv[3], 10)
SAMPLE_INTERVAL = parseInt(process.argv[4], 10)

for doc_and_project_id in process.argv.slice(5)
	do (doc_and_project_id) ->
		[project_id, doc_id] = doc_and_project_id.split(":")
		console.log {project_id, doc_id}
		DocUpdaterClient.setDocLines project_id, doc_id, [(new Array(CLIENT_COUNT + 2)).join('a')], null, null, (error) ->
			throw error if error?
			DocUpdaterClient.getDoc project_id, doc_id, (error, res, body) =>
				throw error if error?
				if !body.lines?
					return console.error "[#{new Date()}] ERROR: Invalid response from get doc (#{doc_id})", body
				content = body.lines.join("\n")
				version = body.version
			
				clients = []
				for pos in [1..CLIENT_COUNT]
					do (pos) ->
						client = new StressTestClient({doc_id, project_id, content, pos: pos, version: version, updateDelay: UPDATE_DELAY})
						clients.push client
				
				do runBatch = () ->
					jobs = clients.map (client) ->
						(cb) -> client.runForNUpdates(SAMPLE_INTERVAL / UPDATE_DELAY, cb)
					async.parallel jobs, (error) ->
						throw error if error?
						printSummary(doc_id, clients)
						checkDocument project_id, doc_id, clients, (error) ->
							throw error if error?
							runBatch()