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

let count = 0

// Handle or rethrow exceptions based on exception properties.
const rescue = require('rescue')

// Return the first non-`null`-like value.
const coalesce = require('extant')

const Trampoline = require('reciprocate')

// Sort function generator.
const ascension = require('ascension')
// Extract the sorted field from an object.
const whittle = require('whittle')

const Interrupt = require('interrupt')

// Strata b-tree iteration ultilities.
const mvcc = {
    // Filter for the latest version of an MVCC record.
    designate: require('designate'),
    // Filter iterator results with a filter function.
    dilute: require('dilute'),
    // Merge two or more iterators.
    homogenize: require('homogenize'),
    // Iterate through a Strata b-tree.
    riffle: require('riffle'),
    // Retrieve a set of values from a Strata b-tree.
    skip: require('skip'),
    // Convert iterator results with a conversion function.
    twiddle: require('twiddle'),
    // Splice a versioned staging tree into a primary tree.
    splice: require('splice')
}

// A `async`/`await` durable b-tree.
const Strata = require('b-tree')
const FileSystem = require('b-tree/filesystem')
const WriteAheadOnly = require('b-tree/writeahead')
const Magazine = require('magazine')
const Fracture = require('fracture')

// Reference counts are kept in an array where the reference count is indexed by
// the position of the stage in the stage array at the time the reference was
// taken. We can merge a stage into the primary tree when it is no longer in the
// zero position of the stage array and the reference count in the zero position
// of the reference array reaches zero. Obviously, the reference count in the
// zero position of the reference array will not increase from zero after the
// has been shifted out of the zero position of the stage array.

// The reference count is not based on readers or writers, but the position.
// Writers do not want the zero position stage to be merged because they are
// still writing to it.

// Readers do not want the zero position stage to be merged because they may be
// reading only a subset of the versions for that stage. When the stage
// is merged the version will be changed to zero and unless they stage has a
// version that overrides the zero, the new value will be returned from a
// reader, ruining the isolation.

// We need three stages, rotate, amalgamate and unstage so we can go from the
// kick-off to waiting for writers to drain to waiting for readers to drain.

// We no longer assert that there are only two stages. We instead amalgamate a
// homogenized set of stages, all the stages after the rotated stage, so that
// once that succeeds we can eliminate all of the old logs and old stages.

//
class Amalgamator {
    static Error = Interrupt.create('Amalgamator.Error', {
        DOES_NOT_EXIST: 'database does not exist',
        NOT_A_DATABASE: 'database directory does not contain a database',
        ALREADY_EXISTS: 'attempted to create a database where one already exists',
        NOT_SAME_STAGE: 'destructible and destructible of turnstile must be in the same shutdown stage'
    })

    constructor (destructible, rotator, open, options) {
        // Implement the Destructible deferrable pattern.
        this.destructible = destructible

        this.deferrable = destructible.durable($ => $(), { countdown: 1 }, 'deferrable')
        this.destructible.destruct(() => this.deferrable.decrement())

        this._destructible = {
            strata: this.destructible.durable($ => $(), 'strata'),
            amalgamate: this.deferrable.durable($ => $(), { countdown: 1 }, 'xxxamalgamate'),
            unstage: this.deferrable.durable($ => $(), { countdown: 1 }, 'unstage')
        }

        this.rotator = rotator

        // Whether or not this Amalgamator should check for conflicts.
        this._conflictable = coalesce(options.conflictable, true)

        // For staging we wrap the application key comparator in a comparator
        // that will include the version and order of the operation. The order
        // is the order in which we processed the record in a batch.
        const stage = ascension([ options.comparator, [ Number, -1 ], [ Number, -1 ] ])

        this.comparator = {
            primary: options.comparator,
            stage: {
                key: stage,
                item: whittle(stage, item => item.key)
            }
        }

        // Transforms an application operation into an `Amalgamator` operation.
        this._transformer = options.transformer

        Amalgamator.Error.assert(this.destructible.isDestroyedIfDestroyed(options.turnstile.destructible), 'NOT_SAME_STAGE')
        this._turnstile = options.turnstile

        // The Strata b-tree cache to use to store pages.
        this._pages = options.pages

        this._stages = []

        const strata = { stage: null, primary: null }

        // Number of records in a staging tree after which the tree is merged
        // into the primary tree.
        strata.stage = coalesce(strata.stage, {})

        // Primary and staging tree split, merge and vacuum properties.
        strata.primary = coalesce(strata.primary, {})

        strata.stage.leaf = coalesce(strata.stage.leaf, {})
        strata.stage.leaf.split = coalesce(strata.stage.leaf.split, 4096)
        strata.stage.leaf.merge = coalesce(strata.stage.leaf.split, 1024)
        strata.stage.branch = coalesce(strata.stage.branch, {})
        strata.stage.branch.split = coalesce(strata.stage.branch.split, 4096)
        strata.stage.branch.merge = coalesce(strata.stage.branch.split, 1024)

        strata.primary.leaf = coalesce(strata.primary.leaf, {})
        strata.primary.leaf.split = coalesce(strata.primary.leaf.split, 4096)
        strata.primary.leaf.merge = coalesce(strata.primary.leaf.split, 1024)
        strata.primary.branch = coalesce(strata.primary.branch, {})
        strata.primary.branch.split = coalesce(strata.primary.branch.split, 4096)
        strata.primary.branch.merge = coalesce(strata.primary.branch.split, 1024)

        this.strata = strata

        {
            const destructible = this._destructible.strata.durable($ => $(), 'primary')
            const storage = new FileSystem.Writer(destructible.durable($ => $(), 'storage'), open.storage)
            this.primary = new Strata(destructible.durable($ => $(), 'strata'), {
                ...options.primary,
                storage: storage,
                turnstile: options.turnstile,
                pages: options.pages.subordinate(),
                comparator: this.comparator.primary,
                extractor: options.extractor
            })
            this.primary.deferrable.increment()
        }
        for (const stage of open.stages) {
            this._newStage(stage)
            this._stages.push(stage)
        }
        this.deferrable.destruct(() => {
            this._open = false
            destructible.ephemeral('shutdown', async () => {
                await this.rotator.done()
                this._destructible.amalgamate.decrement()
                this._destructible.unstage.decrement()
                for (const strata of [ this.primary ].concat(this._stages.map(stage => stage.strata))) {
                    strata.deferrable.decrement()
                }
                await this.drain()
            })
        })
    }<commit_msg>Launch shutdown with correct destructible.<commit_after>// Node.js API.
const path = require('path')
const fs = require('fs').promises
const assert = require('assert')

let count = 0

// Handle or rethrow exceptions based on exception properties.
const rescue = require('rescue')

// Return the first non-`null`-like value.
const coalesce = require('extant')

const Trampoline = require('reciprocate')

// Sort function generator.
const ascension = require('ascension')
// Extract the sorted field from an object.
const whittle = require('whittle')

const Interrupt = require('interrupt')

// Strata b-tree iteration ultilities.
const mvcc = {
    // Filter for the latest version of an MVCC record.
    designate: require('designate'),
    // Filter iterator results with a filter function.
    dilute: require('dilute'),
    // Merge two or more iterators.
    homogenize: require('homogenize'),
    // Iterate through a Strata b-tree.
    riffle: require('riffle'),
    // Retrieve a set of values from a Strata b-tree.
    skip: require('skip'),
    // Convert iterator results with a conversion function.
    twiddle: require('twiddle'),
    // Splice a versioned staging tree into a primary tree.
    splice: require('splice')
}

// A `async`/`await` durable b-tree.
const Strata = require('b-tree')
const FileSystem = require('b-tree/filesystem')
const WriteAheadOnly = require('b-tree/writeahead')
const Magazine = require('magazine')
const Fracture = require('fracture')

// Reference counts are kept in an array where the reference count is indexed by
// the position of the stage in the stage array at the time the reference was
// taken. We can merge a stage into the primary tree when it is no longer in the
// zero position of the stage array and the reference count in the zero position
// of the reference array reaches zero. Obviously, the reference count in the
// zero position of the reference array will not increase from zero after the
// has been shifted out of the zero position of the stage array.

// The reference count is not based on readers or writers, but the position.
// Writers do not want the zero position stage to be merged because they are
// still writing to it.

// Readers do not want the zero position stage to be merged because they may be
// reading only a subset of the versions for that stage. When the stage
// is merged the version will be changed to zero and unless they stage has a
// version that overrides the zero, the new value will be returned from a
// reader, ruining the isolation.

// We need three stages, rotate, amalgamate and unstage so we can go from the
// kick-off to waiting for writers to drain to waiting for readers to drain.

// We no longer assert that there are only two stages. We instead amalgamate a
// homogenized set of stages, all the stages after the rotated stage, so that
// once that succeeds we can eliminate all of the old logs and old stages.

//
class Amalgamator {
    static Error = Interrupt.create('Amalgamator.Error', {
        DOES_NOT_EXIST: 'database does not exist',
        NOT_A_DATABASE: 'database directory does not contain a database',
        ALREADY_EXISTS: 'attempted to create a database where one already exists',
        NOT_SAME_STAGE: 'destructible and destructible of turnstile must be in the same shutdown stage'
    })

    constructor (destructible, rotator, open, options) {
        // Implement the Destructible deferrable pattern.
        this.destructible = destructible

        this.deferrable = destructible.durable($ => $(), { countdown: 1 }, 'deferrable')
        this.destructible.destruct(() => this.deferrable.decrement())

        this._destructible = {
            strata: this.destructible.durable($ => $(), 'strata'),
            amalgamate: this.deferrable.durable($ => $(), { countdown: 1 }, 'xxxamalgamate'),
            unstage: this.deferrable.durable($ => $(), { countdown: 1 }, 'unstage')
        }

        this.rotator = rotator

        // Whether or not this Amalgamator should check for conflicts.
        this._conflictable = coalesce(options.conflictable, true)

        // For staging we wrap the application key comparator in a comparator
        // that will include the version and order of the operation. The order
        // is the order in which we processed the record in a batch.
        const stage = ascension([ options.comparator, [ Number, -1 ], [ Number, -1 ] ])

        this.comparator = {
            primary: options.comparator,
            stage: {
                key: stage,
                item: whittle(stage, item => item.key)
            }
        }

        // Transforms an application operation into an `Amalgamator` operation.
        this._transformer = options.transformer

        Amalgamator.Error.assert(this.destructible.isDestroyedIfDestroyed(options.turnstile.destructible), 'NOT_SAME_STAGE')
        this._turnstile = options.turnstile

        // The Strata b-tree cache to use to store pages.
        this._pages = options.pages

        this._stages = []

        const strata = { stage: null, primary: null }

        // Number of records in a staging tree after which the tree is merged
        // into the primary tree.
        strata.stage = coalesce(strata.stage, {})

        // Primary and staging tree split, merge and vacuum properties.
        strata.primary = coalesce(strata.primary, {})

        strata.stage.leaf = coalesce(strata.stage.leaf, {})
        strata.stage.leaf.split = coalesce(strata.stage.leaf.split, 4096)
        strata.stage.leaf.merge = coalesce(strata.stage.leaf.split, 1024)
        strata.stage.branch = coalesce(strata.stage.branch, {})
        strata.stage.branch.split = coalesce(strata.stage.branch.split, 4096)
        strata.stage.branch.merge = coalesce(strata.stage.branch.split, 1024)

        strata.primary.leaf = coalesce(strata.primary.leaf, {})
        strata.primary.leaf.split = coalesce(strata.primary.leaf.split, 4096)
        strata.primary.leaf.merge = coalesce(strata.primary.leaf.split, 1024)
        strata.primary.branch = coalesce(strata.primary.branch, {})
        strata.primary.branch.split = coalesce(strata.primary.branch.split, 4096)
        strata.primary.branch.merge = coalesce(strata.primary.branch.split, 1024)

        this.strata = strata

        {
            const destructible = this._destructible.strata.durable($ => $(), 'primary')
            const storage = new FileSystem.Writer(destructible.durable($ => $(), 'storage'), open.storage)
            this.primary = new Strata(destructible.durable($ => $(), 'strata'), {
                ...options.primary,
                storage: storage,
                turnstile: options.turnstile,
                pages: options.pages.subordinate(),
                comparator: this.comparator.primary,
                extractor: options.extractor
            })
            this.primary.deferrable.increment()
        }
        for (const stage of open.stages) {
            this._newStage(stage)
            this._stages.push(stage)
        }
        this.deferrable.destruct(() => {
            this._open = false
            this.deferrable.ephemeral('shutdown', async () => {
                await this.rotator.drain()
                this._destructible.amalgamate.decrement()
                this._destructible.unstage.decrement()
                for (const strata of [ this.primary ].concat(this._stages.map(stage => stage.strata))) {
                    strata.deferrable.decrement()
                }
                await this.drain()
            })
        })
    }
{'fraction_non_alphanumeric': 0.10090590491304635, 'fraction_numerical': 0.005025457911789989, 'mean_word_length': 2.619047619047619, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4349755', 'n_tokens_mistral': 4297, 'n_tokens_neox': 4056, 'n_words': 1386}