lib/cache.js
function _writeback(immediate_file, cb) {
    assert.optionalString(immediate_file, 'immediate_file');
    assert.optionalFunc(cb, 'callback');

    var log = this.log;
    var self = this;

    function _doUpload(_arg, _cb) {
        var cachefile = path.join(self.location, 'fscache', _arg.cachefile);
        log.debug('writeback %s => %s', _arg.cachefile, _arg.fname);

        fs.stat(cachefile, function onStatDone(s_err, stats) {
            if (s_err) {
                // This should never happen
                log.error(s_err, '_doUpload(%s): error stat-ing local ' +
                    'file (%s)', _arg.fname, cachefile);
                _cb(s_err, _arg);
                return;
            }

            // Only read the current size in case the file gets extended
            // while we're uploading. In that case it will be dirty again
            // and be re-uploaded.
            var rd_size = stats.size;
            if (rd_size > 0)
                rd_size -= 1;
            var r_opts = {
                start: 0,
                end: rd_size
            };
            var m_opts = {
                size: stats.size
            };
            var rstream = fs.createReadStream(cachefile, r_opts);
            rstream.once('error', function onFileError(err) {
                log.error(err, '_doUpload(%s): error reading from local ' +
                    'file (%s)', _arg.fname, cachefile);
                _cb(err, _arg);
            });

            rstream.once('open', function onOpen() {
                self.manta.put(manta.path(_arg.fname, true), rstream, m_opts,
                  function (err) {
                    if (err) {
                        log.error(err, '_doUpload(%s): error pushing (%s) ' +
                            'to manta:', _arg.fname, cachefile);
                        _cb(err, _arg);
                    } else {
                        log.debug('_doUpload(%s) pushed to (%s):',
                            cachefile, _arg.fname);
                        _cb(null, _arg);
                    }
                });
            });
        });
    }

    function _fileUploaded(_arg, _cb) {
        if (!wb_pending[_arg.fname]) {
            // if no more writes while we were uploading, we remove it from
            // the DB since its no longer dirty
            var key = sprintf(DIRTY_KEY_FMT, _arg.fname);
            self.db.batch()
                .del(key)
                .write(function onBatchDel(err) {
                    if (err) {
                        log.warn(err, 'dirty del(%s): failed', _arg.fname);
                        _cb(errno(err, _arg));
                    } else {
                        log.trace('dirty del(%s): done', _arg.fname);
                        _cb(null, _arg);
                    }
                });
        }
    }

    function _updateLastMod(_arg, _cb) {
        // Update the last modified time for this file into the cache.
        // Ignore errors here since this can only impact our view of when the
        // file is stale.
        self.get(_arg.fname, function (err, info) {
            if (err) {
                _cb(null, _arg);
                return;
            }

            var upd = new Date();
            info.headers['last-modified'] = upd.toUTCString();
            info.last_stat = upd.getTime();

            self.put(_arg.fname, info, function (err2) {
                _cb(null, _arg);
            });
        });
    }

    function _wb_done(err, res) {
        // just use the return from the first operation to get the args
        var args = res.operations[0].result;
        if (err) {
            log.error(err, 'writeback: error %s', args.fname);
            // XXX JJ generalize error handling
            // e.g. if (err.code === 'UploadTimeout') ...
            // other manta errors, etc.
            // On upload timeout we won't have called the _fileUploaded
            // function in the pipeline, so we still have the file in the DB,
            // but not the wb_pending list. We'll add this entry back into the
            // in-memory wb_pending list (or overwrite the existing entry if
            // the file was updated while we tried to upload), but with a
            // timestamp one minute in the future, so that we'll retry the
            // upload of this file a little later. We still keep trying to
            // upload any other files in the list. This handles the case in
            // which the failure was somehow relevant only to this file. If all
            // uploads fail, we'll have just added them back to the writeback
            // list for future retries.
            var plusminute = new Date().getTime();
            plusminute += 60000;
            wb_pending[args.fname] = {
                cachefile: args.cachefile,
                timestamp: plusminute
            };
        } else {
            log.debug('writeback %s done', args.fname);
        }

        var pending_op = wb_inprogress[args.fname].op;
        var pending_cb = wb_inprogress[args.fname].cb;
        var pending_info = wb_inprogress[args.fname].info;
        // remove it from the in-memory inprogress list
        delete wb_inprogress[args.fname];
        wb_num--;

        if (pending_op) {
            log.debug('writeback %s, pending %s', args.fname, pending_op);

            // We had to block an operation while the upload was in progress.
            // Perform that operation now and schedule _writeback to continue.
            setTimeout(_writeback.bind(self), 1);
            if (pending_op === 'del') {
                self.del(args.fname, pending_cb);
            } else if (pending_op === 'trunc') {
                self.truncate(args.fname, pending_info, pending_cb);
            } else if (pending_op === 'cb') {
                pending_cb();
            } else {
                log.error('writeback %s, pending %s unknown',
                    args.fname, pending_op);
            }
        } else {
            // this is the normal path, just take another lap
            var wb = _writeback.bind(self);
            wb();
        }
    }

    if (immediate_file) {
        if (wb_pending[immediate_file]) {
            // the file is dirty, upload it now
            log.debug('_writeback immd. %s: pending', immediate_file);

            file_data = {
                fname: immediate_file,
                cachefile: wb_pending[immediate_file].cachefile
            };

            // remove it from the in-memory list
            delete wb_pending[immediate_file];

            // add it to the in-memory in progress list with cb pending
            wb_inprogress[immediate_file] = {
                op: 'cb',
                cb: cb
            };
            wb_num++;

            // here is where we setup the writeback pipeline
            vasync.pipeline({
                funcs: [
                    _doUpload,
                    _fileUploaded,
                    _updateLastMod
                ],
                arg: file_data
            }, _wb_done);

        } else if (wb_inprogress[immediate_file]) {
            // the file is already being uploaded now, setup cb
            if (wb_inprogress[immediate_file].op) {
                // there shouldn't be any pending op in this case
                log.error('writeback immd. %s: already pending %s',
                    immediate_file, wb_inprogress[immediate_file].op);
            }
            log.debug('_writeback immd. %s: in progress', immediate_file);
            wb_inprogress[immediate_file] = {
                op: 'cb',
                cb: cb
            };
        } else {
            // the file is not dirty and not being uploaded, just run cb
            log.debug('_writeback immd. %s: do nothing', immediate_file);
            cb(null);
        }

        return;
    }

    var now = new Date().getTime();
    var threshold = now - self.wbtime;

    // When we iterate using 'in', the elements in wb_pending are probably
    // ordered by the order in which they were defined. On startup, when
    // pulling out of the DB, we define them in the order we get them from the
    // DB which, since its a btree, will be alphabetical. Thus, we have to look
    // at each entry to see if its time to write it back. That is, the entries
    // might not be in timestamp order so we can't stop when we see an entry
    // that's not ready to be written back.
    //
    // Once we start processing the list, we'll get called back and keep making
    // a pass over the list until we have made a full pass with no writeback.
    // At that time, if there are still elements in the list, we schedule
    // ourselves to wakeup later and process the list again.

    var wb_running = false;
    var more_files = false;
    var fname;
    var file_data = {};
    for (fname in wb_pending) {
        if (threshold > wb_pending[fname].timestamp) {
            if (wb_inprogress[fname]) {
                // This file is still in the middle of an upload and its dirty
                // again and time to writeback again. We skip this file until
                // the current upload finishes and we come back around.
                continue;
            }

            file_data = {
                fname: fname,
                cachefile: wb_pending[fname].cachefile
            };

            // remove it from the in-memory list
            delete wb_pending[fname];

            // add it to the in-memory in progress list with nothing pending
            wb_inprogress[fname] = {};
            wb_num++;
            wb_running = true;

            // here is where we setup the writeback pipeline
            vasync.pipeline({
                funcs: [
                    _doUpload,
                    _fileUploaded,
                    _updateLastMod
                ],
                arg: file_data
            }, _wb_done);

            if (wb_num >= self.num_par) {
                // we've hit our limit for parallel writebacks, break out of
                // the loop since we'll callback to _writeback when one of the
                // uploads has finished
                break;
            }
        } else {
            more_files = true;
        }
    }

    if (!wb_running) {
        if (more_files) {
            setTimeout(_writeback.bind(self), self.wbtime);
        } else {
            writeback_enabled = false;
        }
    }
}
function _writeback(immediate_file, cb) {
    assert.optionalString(immediate_file, 'immediate_file');
    assert.optionalFunc(cb, 'callback');

    var log = this.log;
    var self = this;

    function _doUpload(_arg, _cb) {
        var cachefile = path.join(self.location, 'fscache', _arg.cachefile);
        log.debug('writeback %s => %s', _arg.cachefile, _arg.fname);

        fs.stat(cachefile, function onStatDone(s_err, stats) {
            if (s_err) {
                // This should never happen
                log.error(s_err, '_doUpload(%s): error stat-ing local ' +
                    'file (%s)', _arg.fname, cachefile);
                _cb(s_err, _arg);
                return;
            }

            // Only read the current size in case the file gets extended
            // while we're uploading. In that case it will be dirty again
            // and be re-uploaded.
            var rd_size = stats.size;
            if (rd_size > 0)
                rd_size -= 1;
            var r_opts = {
                start: 0,
                end: rd_size
            };
            var m_opts = {
                size: stats.size
            };
            var rstream = fs.createReadStream(cachefile, r_opts);
            rstream.once('error', function onFileError(err) {
                log.error(err, '_doUpload(%s): error reading from local ' +
                    'file (%s)', _arg.fname, cachefile);
                _cb(err, _arg);
            });

            rstream.once('open', function onOpen() {
                self.manta.put(manta.path(_arg.fname, true), rstream, m_opts,
                  function (err) {
                    if (err) {
                        log.error(err, '_doUpload(%s): error pushing (%s) ' +
                            'to manta:', _arg.fname, cachefile);
                        _cb(err, _arg);
                    } else {
                        log.debug('_doUpload(%s) pushed to (%s):',
                            cachefile, _arg.fname);
                        _cb(null, _arg);
                    }
                });
            });
        });
    }

    function _fileUploaded(_arg, _cb) {
        if (!wb_pending[_arg.fname]) {
            // if no more writes while we were uploading, we remove it from
            // the DB since its no longer dirty
            var key = sprintf(DIRTY_KEY_FMT, _arg.fname);
            self.db.batch()
                .del(key)
                .write(function onBatchDel(err) {
                    if (err) {
                        log.warn(err, 'dirty del(%s): failed', _arg.fname);
                        _cb(errno(err, _arg));
                    } else {
                        log.trace('dirty del(%s): done', _arg.fname);
                        _cb(null, _arg);
                    }
                });
        }
    }

    function _updateLastMod(_arg, _cb) {
        // Update the last modified time for this file into the cache.
        // Ignore errors here since this can only impact our view of when the
        // file is stale.
        self.get(_arg.fname, function (err, info) {
            if (err) {
                _cb(null, _arg);
                return;
            }

            var upd = new Date();
            info.headers['last-modified'] = upd.toUTCString();
            info.last_stat = upd.getTime();

            self.put(_arg.fname, info, function (err2) {
                _cb(null, _arg);
            });
        });
    }

    function _wb_done(err, res) {
        // just use the return from the first operation to get the args
        var args = res.operations[0].result;
        if (err) {
            log.error(err, 'writeback: error %s', args.fname);
            // XXX JJ generalize error handling
            // e.g. if (err.code === 'UploadTimeout') ...
            // other manta errors, etc.
            // On upload timeout we won't have called the _fileUploaded
            // function in the pipeline, so we still have the file in the DB,
            // but not the wb_pending list. We'll add this entry back into the
            // in-memory wb_pending list (or overwrite the existing entry if
            // the file was updated while we tried to upload), but with a
            // timestamp one minute in the future, so that we'll retry the
            // upload of this file a little later. We still keep trying to
            // upload any other files in the list. This handles the case in
            // which the failure was somehow relevant only to this file. If all
            // uploads fail, we'll have just added them back to the writeback
            // list for future retries.
            var plusminute = new Date().getTime();
            plusminute += 60000;
            wb_pending[args.fname] = {
                cachefile: args.cachefile,
                timestamp: plusminute
            };
        } else {
            log.debug('writeback %s done', args.fname);
        }

        var pending_op = wb_inprogress[args.fname].op;
        var pending_cb = wb_inprogress[args.fname].cb;
        var pending_info = wb_inprogress[args.fname].info;
        // remove it from the in-memory inprogress list
        delete wb_inprogress[args.fname];
        wb_num--;

        if (pending_op) {
            log.debug('writeback %s, pending %s', args.fname, pending_op);

            // We had to block an operation while the upload was in progress.
            // Perform that operation now and schedule _writeback to continue.
            setTimeout(_writeback.bind(self), 1);
            if (pending_op === 'del') {
                self.del(args.fname, pending_cb);
            } else if (pending_op === 'trunc') {
                self.truncate(args.fname, pending_info, pending_cb);
            } else if (pending_op === 'cb') {
                pending_cb();
            } else {
                log.error('writeback %s, pending %s unknown',
                    args.fname, pending_op);
            }
        } else {
            // this is the normal path, just take another lap
            var wb = _writeback.bind(self);
            wb();
        }
    }

    if (immediate_file) {
        if (wb_pending[immediate_file]) {
            // the file is dirty, upload it now
            log.debug('_writeback immd. %s: pending', immediate_file);

            file_data = {
                fname: immediate_file,
                cachefile: wb_pending[immediate_file].cachefile
            };

            // remove it from the in-memory list
            delete wb_pending[immediate_file];

            // add it to the in-memory in progress list with cb pending
            wb_inprogress[immediate_file] = {
                op: 'cb',
                cb: cb
            };
            wb_num++;

            // here is where we setup the writeback pipeline
            vasync.pipeline({
                funcs: [
                    _doUpload,
                    _fileUploaded,
                    _updateLastMod
                ],
                arg: file_data
            }, _wb_done);

        } else if (wb_inprogress[immediate_file]) {
            // the file is already being uploaded now, setup cb
            if (wb_inprogress[immediate_file].op) {
                // there shouldn't be any pending op in this case
                log.error('writeback immd. %s: already pending %s',
                    immediate_file, wb_inprogress[immediate_file].op);
            }
            log.debug('_writeback immd. %s: in progress', immediate_file);
            wb_inprogress[immediate_file] = {
                op: 'cb',
                cb: cb
            };
        } else {
            // the file is not dirty and not being uploaded, just run cb
            log.debug('_writeback immd. %s: do nothing', immediate_file);
            cb(null);
        }

        return;
    }

    var now = new Date().getTime();
    var threshold = now - self.wbtime;

    // When we iterate using 'in', the elements in wb_pending are probably
    // ordered by the order in which they were defined. On startup, when
    // pulling out of the DB, we define them in the order we get them from the
    // DB which, since its a btree, will be alphabetical. Thus, we have to look
    // at each entry to see if its time to write it back. That is, the entries
    // might not be in timestamp order so we can't stop when we see an entry
    // that's not ready to be written back.
    //
    // Once we start processing the list, we'll get called back and keep making
    // a pass over the list until we have made a full pass with no writeback.
    // At that time, if there are still elements in the list, we schedule
    // ourselves to wakeup later and process the list again.

    var wb_running = false;
    var more_files = false;
    var fname;
    var file_data = {};
    for (fname in wb_pending) {
        if (threshold > wb_pending[fname].timestamp) {
            if (wb_inprogress[fname]) {
                // This file is still in the middle of an upload and its dirty
                // again and time to writeback again. We skip this file until
                // the current upload finishes and we come back around.
                continue;
            }

            file_data = {
                fname: fname,
                cachefile: wb_pending[fname].cachefile
            };

            // remove it from the in-memory list
            delete wb_pending[fname];

            // add it to the in-memory in progress list with nothing pending
            wb_inprogress[fname] = {};
            wb_num++;
            wb_running = true;

            // here is where we setup the writeback pipeline
            vasync.pipeline({
                funcs: [
                    _doUpload,
                    _fileUploaded,
                    _updateLastMod
                ],
                arg: file_data
            }, _wb_done);

            if (wb_num >= self.num_par) {
                // we've hit our limit for parallel writebacks, break out of
                // the loop since we'll callback to _writeback when one of the
                // uploads has finished
                break;
            }
        } else {
            more_files = true;
        }
    }

    if (!wb_running) {
        if (more_files) {
            setTimeout(_writeback.bind(self), self.wbtime);
        } else {
            writeback_enabled = false;
        }
    }
}
This function drives the writeback of dirty files.

Writeback of dirty files is normally asynchronous. It is initiated on
startup if we have dirty files in the DB and its scheduled by the 'dirty'
function if necessary. Dirty files are held in the local cache for a period
of time before being written back.

This function also calls itself when a writeback has completed to see if
there is more work to do.

The _writeback function is normally called with no arguments and simply
processes the list of dirty files to upload the ones which have aged out.
However, if called explicitly with a filename and callback, then it will
immediately upload that file (if its dirty) and then run the callback.