From 8d3e0b2353fd8e378a8a1246c786912523f5d145 Mon Sep 17 00:00:00 2001 From: Brian Gough Date: Thu, 21 May 2015 15:46:29 +0100 Subject: [PATCH] added consistency check to pack migration script --- services/track-changes/pack.coffee | 162 ++++++++++++++++++++--------- 1 file changed, 114 insertions(+), 48 deletions(-) diff --git a/services/track-changes/pack.coffee b/services/track-changes/pack.coffee index 8d31329ffb..27e9a347b8 100644 --- a/services/track-changes/pack.coffee +++ b/services/track-changes/pack.coffee @@ -9,12 +9,94 @@ util = require 'util' _ = require 'underscore' lineReader = require "line-reader" +cli = require "cli" +options = cli.parse { + 'dry-run': ['n', 'do not write to database'], + 'fast': [false, 'no delays on writes'] +} MAX_SIZE = 1024*1024 MAX_COUNT = 1024 MIN_COUNT = 100 KEEP_OPS = 100 +DB_WRITE_DELAY = if options.fast then 0 else 2000 +DOCUMENT_PACK_DELAY = if options.fast then 0 else 1000 + +packDocHistory = (doc_id, callback) -> + util.log "starting pack operation for #{doc_id}" + getDocHistory doc_id, (err, docs) -> + return callback(err) if err? + origDocs = docs.length + convertDocsToPacks docs, (err, packs) -> + return callback(err) if err? + util.log "docs #{origDocs} packs #{packs.length}" + if packs.length + if options['dry-run'] + util.log 'dry-run, skipping write packs' + return callback() + savePacks packs, (err) -> + return callback(err) if err? + # check the history again + getDocHistory doc_id, callback + else + util.log "no packs to write" + callback null, null + +# retrieve document ops/packs and check them +getDocHistory = (doc_id, callback) -> + db.docHistory.find({doc_id:mongojs.ObjectId(doc_id)}).sort {v:1}, (err, docs) -> + return callback(err) if err? + # for safety, do a consistency check of the history + checkHistory doc_id, docs, (err) -> + return callback(err) if err? + callback err, docs + +convertDocsToPacks = (docs, callback) -> + packs = [] + top = null + # keep the last KEEP_OPS as individual ops + docs = docs.slice(0,-KEEP_OPS) + + docs.forEach (d,i) -> + if d.pack? + util.log "skipping existing pack of #{d.pack.length}" + top = null if top? # flush the current pack + return # and try next + sz = BSON.calculateObjectSize(d) + if top? && top.pack.length < MAX_COUNT && top.sz + sz < MAX_SIZE + top.pack = top.pack.concat {v: d.v, meta: d.meta, op: d.op, _id: d._id} + top.sz += sz + top.v_end = d.v + top.meta.end_ts = d.meta.end_ts + return + else if sz < MAX_SIZE + # create a new pack + top = _.clone(d) + top.pack = [ {v: d.v, meta: d.meta, op: d.op, _id: d._id} ] + top.meta = { start_ts: d.meta.start_ts, end_ts: d.meta.end_ts } + top.sz = sz + delete top.op + delete top._id + packs.push top + else + # keep the op + util.log "keeping large op unchanged (#{sz} bytes)" + + # only store packs with a sufficient number of ops, discard others + packs = packs.filter (packObj) -> + packObj.pack.length > MIN_COUNT + callback(null, packs) + +savePacks = (packs, callback) -> + async.eachSeries packs, insertPack, (err, result) -> + if err? + console.log err + callback err, result + else + util.log "done writing packs" + callback() + insertPack = (packObj, callback) -> if shutdownRequested return callback('shutdown') @@ -31,52 +113,34 @@ insertPack = (packObj, callback) -> console.log 'nInserted', result.nInserted, 'nRemoved', result.nRemoved setTimeout () -> callback(err, result) - , 2000 + , DB_WRITE_DELAY -packDocHistory = (doc_id, callback) -> - util.log "starting pack operation for #{doc_id}" - db.docHistory.find({doc_id:mongojs.ObjectId(doc_id),pack:{$exists:false}}).sort {v:1}, (err, docs) -> - packs = [] - top = null - origDocs = docs.length - # keep the last KEEP_OPS as individual ops - docs = docs.slice(0,-KEEP_OPS) - docs.forEach (d,i) -> - sz = BSON.calculateObjectSize(d) - if top? && top.pack.length < MAX_COUNT && top.sz + sz < MAX_SIZE - top.pack = top.pack.concat {v: d.v, meta: d.meta, op: d.op, _id: d._id} - top.sz += sz - top.v_end = d.v - top.meta.end_ts = d.meta.end_ts - return - else if sz < MAX_SIZE - # create a new pack - top = _.clone(d) - top.pack = [ {v: d.v, meta: d.meta, op: d.op, _id: d._id} ] - top.meta = { start_ts: d.meta.start_ts, end_ts: d.meta.end_ts } - top.sz = sz - delete top.op - delete top._id - packs.push top - else - # keep the op - util.log 'keeping large op unchanged (#{sz} bytes)' - - # only store packs with a sufficient number of ops, discard others - packs = packs.filter (packObj) -> - packObj.pack.length > MIN_COUNT - - util.log "docs #{origDocs} packs #{packs.length}" - if packs.length - async.eachSeries packs, insertPack, (err, result) -> - if err? - console.log doc_id, err - else - util.log "done writing packs" - callback err, result +checkHistory = (doc_id, docs, callback) -> + util.log "checking history for #{doc_id}" + errors = 0 + prev = null + error = (args...) -> + errors++ + console.log.apply(null, args) + docs.forEach (d,i) -> + if d.pack? + n = d.pack.length + last = d.pack[n-1] + error('bad pack v_end', d) if d.v_end != last.v + error('bad pack start_ts', d) if d.meta.start_ts != d.pack[0].meta.start_ts + error('bad pack end_ts', d) if d.meta.end_ts != last.meta.end_ts + d.pack.forEach (p, i) -> + prev = v + v = p.v + error('bad version', v, 'in', p) if v <= prev else - util.log "no packs to write" - callback null, null + prev = v + v = d.v + error('bad version', v, 'in', d) if v <= prev + if errors + callback({errcount: errors}) + else + callback() readFile = (file, callback) -> ids = [] @@ -87,8 +151,10 @@ readFile = (file, callback) -> .then () -> callback(null, ids) -todoFile = process.argv[2] -doneFile = process.argv[3] +todoFile = cli.args[1] +doneFile = cli.args[2] +util.log "reading from #{todoFile}" +util.log "logging progress to #{doneFile}" fs.appendFileSync doneFile, '# starting pack run at ' + new Date() + '\n' shutdownRequested = false @@ -103,13 +169,13 @@ readFile todoFile, (err, todo) -> packDocHistory doc_id, (err, result) -> if err? return callback(err) - else + else if not options['dry-run'] fs.appendFileSync doneFile, doc_id + '\n' if shutdownRequested return callback('shutdown') setTimeout () -> callback(err, result) - , 1000 + , DOCUMENT_PACK_DELAY , (err, results) -> if err? console.log 'error:', err