writeahead.js

// # WriteAhead

// A write-ahead only storage strategy for Strata. We use the `writeahead`
// module to perform synchronous writes to the in-memory `writeahead` queue
// which will be occasionally flushed by a background thread.

//
'use strict'
//

// Node.js API.
const assert = require('assert')
const fs = require('fs').promises
const path = require('path')
//

// Buffer serialization.
const { Player, Recorder } = require('transcript')
//

// Strata modules.
const Strata = { Error: require('./error') }
const Storage = require('./storage')
//

const Fracture = require('fracture')

// Convert an array of records into a single serialized buffer.

//
function _recordify (recorder, records) {
    return Buffer.concat(records.map(record => {
        return recorder([[ Buffer.from(JSON.stringify(record.header)) ].concat(record.parts || [])])
    }))
}
//

// Our write-ahead log associates a block with a set of keys. These keys are
// used to find blocks within the write-ahead log and return them as a series of
// buffers as if they were a single file. We'll store a series of log update
// records in the blocks. Because the records in the series may apply apply to
// different pages, group the records by the page id they apply to and prepend
// an `'apply'` record that indicates two which page the following records
// apply.

// We have two iterators. An `async` iterator that returns each record we
// extract from the log and one that returns invokes a callback with last record
// we extract from the log.

//
const wal = {
    async *iterator (writeahead, qualifier, key) {
        const player = new Player(() => '0')
        let apply
        const entries = []
        for await (const block of writeahead.get([ qualifier, key ])) {
           for (const entry of player.split(block)) {
                const header = JSON.parse(String(entry.parts.shift()))
                if (header.method == 'apply') {
                    apply = header.key == key
                } else if (apply) {
                    entries.push({ header, parts: entry.parts, sizes: entry.sizes })
                }
            }
            if (entries.length != 0) {
                yield entries
                entries.length = 0
            }
        }
        if (entries.length != 0) {
            yield entries
        }
    },
    async last (writeahead, qualifier, key, last = null) {
        const player = new Player(() => '0')
        let apply = false
        for await (const block of writeahead.get([ qualifier, key ])) {
            for (const entry of player.split(block)) {
                const header = JSON.parse(String(entry.parts.shift()))
                if (header.method == 'apply') {
                    apply = header.key == key
                } else if (apply) {
                    last = header
                }
            }
        }
        return last
    },
    async last2 (writeahead, qualifier, key, missing = null) {
        const player = new Player(() => '0')
        let apply = false
        let last = null
        for await (const block of writeahead.get([ qualifier, key ])) {
            for (const entry of player.split(block)) {
                const header = JSON.parse(String(entry.parts.shift()))
                if (header.method == 'apply') {
                    apply = header.key == key
                } else if (apply) {
                    last = { header, parts: entry.parts }
                }
            }
        }
        return last || missing
    }
}

class WriteAheadOnly {
    static wal = wal

    static async open (stack, options) {
        assert(stack instanceof Fracture.Stack)
        options = Storage.options(options)
        const recorder = Recorder.create(() => '0')
        if (options.create) {
            const { create, key, writeahead } = options
            await options.writeahead.write(stack, [{
                keys: [[ key, '0.0' ], [ key, 'instance' ], create ],
                buffer: _recordify(recorder, [{
                    header: {
                        method: 'apply',
                        key: 'instance'
                    }
                }, {
                    header: {
                        method: 'instance',
                        instance: 0
                    }
                }, {
                    header: {
                        method: 'apply',
                        key: '0.0'
                    }
                }, {
                    header: {
                        method: 'insert',
                        index: 0,
                        id: '0.1'
                    }
                }, {
                    header: {
                        method: 'apply',
                        key: create[1]
                    }
                }, {
                    header: {
                        method: 'locate',
                        value: key
                    }
                }])
            }])
            return { ...options, instance: 0, pageId: 2, recorder, pageId: 2 }
        }
        const { key, writeahead } = options
        const instance = (await wal.last2(writeahead, key, 'instance')).header.instance + 1
        await options.writeahead.write(stack, [{
            keys: [[ key, 'instance' ]],
            buffer: _recordify(recorder, [{
                header: {
                    method: 'apply',
                    key: 'instance'
                }
            }, {
                header: {
                    method: 'instance',
                    instance: instance
                }
            }])
        }])
        return { ...options, instance, pageId: 0, recorder, pageId: 0 }
    }

    static Serializer = class {
        constructor (writeahead, qualifier) {
            this._writeahead = writeahead
            this._qualifier = qualifier
            this._records = {}
        }
        push (key, ...records) {
            if (this._records[key] == null) {
                this._records[key] = [{
                    header: { method: 'apply', key: key }
                }]
            }
            this._records[key].push.apply(this._records[key], records)
        }
        get body () {
            const body = []
            for (const key in this._records) {
                body.push.apply(body, this._records[key])
            }
            return body
        }
        get keys () {
            return Object.keys(this._records).map(key => [ this._qualifier, key ])
        }
        serialize () {
            return {
                keys: this.keys,
                buffer: _recordify(this._writeahead._recorder, this.body)
            }
        }

    }

    static Reader = class {
        called = 0

        constructor (options) {
            options = Storage.options(options)
            this.writeahead = options.writeahead
            this.key = options.key
            this.serializer = options.serializer
            this.extractor = options.extractor
            this.checksum = options.checksum
        }

        page (id) {
            return this.log(id, null)
        }

        async log (id, stop) {
            this.called++
            const leaf = +id.split('.')[1] % 2 == 1
            const page = leaf ? {
                id: id,
                items: [],
                right: null,
                key: null,
                stop: 0,
                leaf: true
            } : {
                id: id,
                items: [],
                stop: 0,
                leaf: false
            }
            let apply = false
            const player = new Player(() => '0')
            WAL: for await (const entries of wal.iterator(this.writeahead, this.key, id)) {
                for (const { header, parts, sizes } of entries) {
                    switch (header.method) {
                    case 'stop': {
                            assert(! isNaN(header.stop))
                            page.stop = header.stop
                            if (page.stop == stop) {
                                break WAL
                            }
                            page.stop++
                            assert(! isNaN(header.stop))
                        }
                        break
                    case 'clear': {
                            page.items = []
                        }
                        break
                    case 'load': {
                            const { page: load } = await this.log(header.id, header.stop)
                            page.items = load.items
                            page.stop = load.stop + 1
                            if (leaf) {
                                page.key = load.key
                                page.right = load.right
                            }
                        }
                        break
                    case 'split': {
                            page.items = page.items.slice(header.index, header.length)
                            if (!leaf) {
                                page.items[0].key = null
                            }
                        }
                        break
                    case 'merge': {
                            const { page: load } = await this.log(header.id)
                            if (leaf) {
                                page.right = load.right
                            } else {
                                load.items[0].key = this.serializer.key.deserialize(parts)
                            }
                            page.items.push.apply(page.items, load.items)
                        }
                        break
                    case 'key': {
                            page.key = this.serializer.key.deserialize(parts)
                        }
                        break
                    case 'right': {
                            page.right = this.serializer.key.deserialize(parts)
                        }
                        break
                    case 'insert': {
                            const heft = sizes.reduce((sum, size) => sum + size, 0)
                            if (leaf) {
                                const deserialized = this.serializer.parts.deserialize(parts)
                                page.items.splice(header.index, 0, {
                                    key: this.extractor(deserialized),
                                    parts: deserialized,
                                    heft: heft
                                })
                            } else {
                                const key = parts.length != 0
                                    ? this.serializer.key.deserialize(parts)
                                    : null
                                page.items.splice(header.index, 0, {
                                    id: header.id,
                                    key: key,
                                    heft: heft
                                })
                            }
                        }
                        break
                    case 'delete': {
                            page.items.splice(header.index, 1)
                        }
                        break
                    }
                }
            }
            const heft = page.items.reduce((sum, item) => sum + item.heft, 0)
            return { page, heft }
        }
    }

    static Writer = class {
        constructor (destructible, { writeahead, key, recorder, extractor, serializer, instance, pageId }) {
            this.destructible = destructible
            this.deferrable = destructible.durable($ => $(), { countdown: 1 }, 'deferrable')
            this.destructible.destruct(() => this.deferrable.decrement())
            this._writeahead = writeahead
            this._writeahead.deferrable.increment()
            this.deferrable.destruct(() => this._writeahead.deferrable.decrement())
            this._key = key
            this._id = 0
            this._pageId = pageId
            this.instance = instance
            this.extractor = extractor
            this.serializer = serializer
            this._recorder = recorder
            this.reader = new WriteAheadOnly.Reader({ writeahead, key, extractor, serializer })
        }

        recordify (header, parts = []) {
            return this._recorder([[ Buffer.from(JSON.stringify(header)) ].concat(parts)])
        }

        nextId (leaf) {
            let id
            do {
                id = this._pageId++
            } while (leaf ? id % 2 == 0 : id % 2 == 1)
            return String(this.instance) + '.' +  String(id)
        }

        read (id) {
            return this.reader.page(id)
        }

        writeLeaf (stack, page, writes) {
            writes.unshift(_recordify(this._recorder, [{ header: { method: 'apply', key: page.id } }]))
            this._writeahead.write(stack, [{ keys: [[ this._key, page.id ]], buffer: Buffer.concat(writes) }])
        }

        _serializeMessages (messages) {
            console.log(messages)
            return messages
        }

        //
        _startBalance (serializer, messages) {
            const id = [ 'balance', this.instance, this._id++ ].join('.')

            serializer.push('balance', {
                header: {
                    method: 'balance',
                    key: id
                }
            })

            const serialized = []
            this._serialize(messages, serialized)

            serializer.push(id, {
                header: {
                    method: 'messages',
                    messages: messages
                },
                parts: serialized
            })

            this._write = serializer
        }

        // Even more approximate than usual because we're not accounting for
        // keys that were set to `null` or set to a value from `null`.

        //
        _setHeft (...branches) {
            for (const branch of branches) {
                branch.cartridge.heft = branch.page.items.reduce((sum, item) => sum + item.heft, 0)
            }
        }

        writeDrainRoot ({ left, right, root }) {
            this._setHeft(left, right, root)
            const stop = root.page.stop++
            this._write.push(root.page.id, {
                header: {
                    method: 'stop',
                    stop: stop
                }
            }, {
                header: {
                    method: 'clear'
                }
            }, {
                header: {
                    method: 'insert',
                    index: 0,
                    id: root.page.items[0].id
                }
            }, {
                header: {
                    method: 'insert',
                    index: 1,
                    id: root.page.items[1].id
                },
                parts: this.serializer.key.serialize(root.page.items[1].key)
            })
            this._write.push(left.page.id, {
                header: {
                    method: 'load',
                    id: root.page.id,
                    stop: stop
                }
            }, {
                header: {
                    method: 'split',
                    index: 0,
                    length: left.page.items.length
                }
            })
            this._write.push(right.page.id, {
                header: {
                    method: 'load',
                    id: root.page.id,
                    stop: stop
                }
            }, {
                header: {
                    method: 'split',
                    index: left.page.items.length,
                    length: left.page.items.length + right.page.items.length
                }
            })
        }

        writeSplitBranch ({ promotion, left, right, parent }) {
            this._setHeft(left, right, parent)
            const stop = left.page.stop++
            this._write.push(left.page.id, {
                header: {
                    method: 'stop',
                    stop: stop
                }
            }, {
                header: {
                    method: 'split',
                    index: 0,
                    length: left.page.items.length
                }
            })
            this._write.push(right.page.id, {
                header: {
                    method: 'load',
                    id: left.page.id,
                    stop: stop
                }
            }, {
                header: {
                    method: 'split',
                    index: left.page.items.length,
                    length: left.page.items.length + right.page.items.length
                }
            })
            this._write.push(parent.page.id, {
                header: {
                    method: 'insert',
                    index: parent.index + 1,
                    id: right.page.id
                },
                parts: this.serializer.key.serialize(promotion)
            })

        }

        writeSplitLeaf({ stack, left, right, parent, writes, messages }) {
            this.writeLeaf(stack, left.page, writes)

            const partition = left.page.items.length
            const length = left.page.items.length + right.page.items.length

            const body = []

            const serializer = new WriteAheadOnly.Serializer(this, this._key)

            serializer.push(left.page.id, {
                header: {
                    method: 'stop',
                    stop: left.page.stop++
                }
            }, {
                header: {
                    method: 'split',
                    index: 0,
                    length: partition
                }
            }, {
                header: {
                    method: 'right'
                },
                parts: this.serializer.key.serialize(right.page.key)
            })

            serializer.push(right.page.id, {
                header: {
                    method: 'load',
                    id: left.page.id,
                    stop: left.page.stop - 1
                }
            }, {
                header: {
                    method: 'split',
                    index: partition,
                    length: length
                }
            }, {
                header: {
                    method: 'key'
                },
                parts: this.serializer.key.serialize(right.page.key)
            })

            serializer.push(parent.page.id, {
                header: {
                    method: 'insert',
                    index: parent.index + 1,
                    id: right.page.id
                },
                parts: this.serializer.key.serialize(right.page.items[0].key)
            })

            this._startBalance(serializer, messages)
        }

        writeFillRoot({ root, child }) {
            this._setHeft(root)
            this._write.push(root.page.id, {
                header: {
                    method: 'clear'
                }
            }, {
                header: {
                    method: 'load',
                    id: child.page.id,
                    stop: child.page.stop
                }
            })
        }

        writeMerge ({ key, serializer, left, right, surgery, pivot }) {
            this._setHeft(left, pivot, surgery.splice)
            serializer.push(left.page.id, {
                header: {
                    method: 'merge',
                    id: right.page.id
                },
                parts: left.page.leaf ? [] : this.serializer.key.serialize(key)
            })

            serializer.push(surgery.splice.page.id, {
                header: {
                    method: 'delete',
                    index: surgery.splice.index
                }
            })

            if (surgery.splice.index == 0) {
                serializer.push(surgery.splice.page.id, {
                    header: {
                        method: 'key',
                        index: 0
                    }
                })
            }

            if (surgery.replacement != null) {
                serializer.push(pivot.page.id, {
                    header: {
                        method: 'key',
                        index: pivot.index
                    },
                    parts: this.serializer.key.serialize(surgery.replacement)
                })
            }
        }

        writeMergeBranch ({ key, left, right, surgery, pivot }) {
            this.writeMerge({ key, serializer: this._write, left, right, surgery, pivot })
        }

        writeMergeLeaf ({ stack, left, right, surgery, pivot, writes, messages }) {
            this.writeLeaf(stack, left.page, writes.left)
            this.writeLeaf(stack, right.page, writes.right)

            const serializer = new WriteAheadOnly.Serializer(this, this._key)

            this.writeMerge({ serializer, left, right, surgery, pivot })

            this._startBalance(serializer, messages)
        }

        async balance (stack, sheaf) {
            const cartridges = []
            for (;;) {
                cartridges.splice(0).forEach(cartridge => cartridge.release())
                if (this._write != null) {
                    const write = this._write.serialize()
                    this._write = null
                    await this._writeahead.write(stack, [ write ])
                }
                const { header: balance } = await wal.last2(this._writeahead, this._key, 'balance')
                const { header: { messages }, parts: serialized } = await wal.last2(this._writeahead, this._key, balance.key)
                if (messages.length == 0) {
                    break
                }
                this._write = new WriteAheadOnly.Serializer(this, this._key)
                const message = messages.shift()
                Strata.Error.assert(message.method == 'balance', 'JOURNAL_CORRUPTED')
                const append = []
                switch (message.method) {
                case 'balance':
                    if (message.key == 0) {
                        message.key = null
                    } else {
                        message.key = this.serializer.key.deserialize(serialized.splice(0, message.key))
                    }
                    await sheaf.balance(message.key, message.level, append, cartridges)
                    break
                }
                this._serialize(append, serialized)
                messages.push.apply(messages, append)
                this._write.push(balance.key, {
                    header: {
                        method: 'messages', messages: messages
                    },
                    parts: serialized
                })
            }
        }

        _serialize (append, serialized) {
            for (const message of append) {
                assert.equal(message.method, 'balance')
                if (message.key == null) {
                    message.key = 0
                } else {
                    const parts = this.serializer.key.serialize(message.key)
                    message.key = parts.length
                    serialized.push.apply(serialized, parts)
                }
            }
            return serialized
        }
    }
}

module.exports = WriteAheadOnly