import assert from "assert" ; import { deserialize , serialize } from "v8" ; import { Storage , StoredValue , addAll , runWithInputGateClosed , viewToArray , waitUntilOnOutputGate , } from "@miniflare/shared" ; import { DurableObjectError } from "./error" ; import { ReadWriteMutex } from "./rwmutex" ; import { ShadowStorage } from "./shadow" ; const MAX_KEYS = 128 ; const MAX_KEY_SIZE = 2048 ; const MAX_VALUE_SIZE = 128 * 1024 ; const ENFORCED_MAX_VALUE_SIZE = MAX_VALUE_SIZE + 32 ; const undefinedKeyError = ": parameter 1 is not of type 'variant'. (key is undefined)" ; function intersects < T > ( a , b ) { for ( const value of a ) if ( b . has ( value ) ) return true ; return false ; } export interface DurableObjectGetOptions { allowConcurrency ? : boolean ; noCache ? : boolean ; } export interface DurableObjectPutOptions extends DurableObjectGetOptions { allowUnconfirmed ? : boolean ; } export interface DurableObjectListOptions extends DurableObjectGetOptions { start ? : string ; end ? : string ; prefix ? : string ; reverse ? : boolean ; limit ? : number ; } export interface DurableObjectOperator { get < Value = unknown > ( key , options ? ) : Promise < Value | undefined > ; get < Value = unknown > ( keys , options ? ) : Promise < Map < string , Value > > ; put < Value = unknown > ( key , value , options ? ) : Promise < void > ; put < Value = unknown > ( entries , options ? ) : Promise < void > ; delete ( key , options ? ) : Promise < boolean > ; delete ( keys , options ? ) : Promise < number > ; list < Value = unknown > ( options ? ) : Promise < Map < string , Value > > ; } function assertKeySize ( key , many = false ) { if ( Buffer . byteLength ( key ) <= MAX_KEY_SIZE ) return ; if ( many ) { throw new RangeError ( `Key "${ key }" is larger than the limit of ${ MAX_KEY_SIZE } bytes.` ) ; } throw new RangeError ( `Keys cannot be larger than ${ MAX_KEY_SIZE } bytes.` ) ; } function assertValueSize ( value , key ? ) { if ( value . byteLength <= ENFORCED_MAX_VALUE_SIZE ) return ; if ( key !== undefined ) { throw new RangeError ( `Value for key "${ key }" is above the limit of ${ MAX_VALUE_SIZE } bytes.` ) ; } throw new RangeError ( `Values cannot be larger than ${ MAX_VALUE_SIZE } bytes.` ) ; } function helpfulDeserialize ( buffer ) { try { return deserialize ( buffer ) ; } catch ( e ) { throw new DurableObjectError ( "ERR_DESERIALIZATION" , "Unable to deserialize stored Durable Object data due to an " + "invalid or unsupported version.\nThe Durable Object data storage " + "format changed in Miniflare 2. You cannot load Durable Object data " + "created with Miniflare 1 and must delete it." , e ) ; } } async function get < Value = unknown > ( storage , key , checkMaxKeys ? ) ; async function get < Value = unknown > ( storage , keys , checkMaxKeys ? ) ; async function get < Value = unknown > ( storage , keys , checkMaxKeys = true ) { if ( Array . isArray ( keys ) ) { if ( checkMaxKeys && keys . length > MAX_KEYS ) { throw new RangeError ( `Maximum number of keys is ${ MAX_KEYS }.` ) ; } const defined = [ ] ; for ( const key of keys ) { if ( key === undefined ) continue ; defined . push ( key ) ; assertKeySize ( key , true ) ; } const res = new Map < string , Value > ( ) ; const values = await storage . getMany ( defined ) ; assert . strictEqual ( defined . length , values . length ) ; for ( let i = 0 ; i < defined . length ; i ++ ) { const value = values [ i ] ; if ( value !== undefined ) { res . set ( defined [ i ] , helpfulDeserialize ( value . value ) ) ; } } return res ; } assertKeySize ( keys ) ; const value = await storage . get ( keys ) ; return value && helpfulDeserialize ( value . value ) ; } async function list < Value = unknown > ( storage , options = { } ) { if ( options . limit !== undefined && options . limit <= 0 ) { throw new TypeError ( "List limit must be positive." ) ; } const { keys } = await storage . list ( options ) ; return get ( storage , keys . map ( ( { name } ) => name ) , false ) ; } function normalisePutEntries < Value = unknown > ( keyEntries , valueOptions ? ) { if ( typeof keyEntries === "string" ) { assertKeySize ( keyEntries ) ; if ( valueOptions === undefined ) { throw new TypeError ( "put() called with undefined value." ) ; } const serialized = serialize ( valueOptions ) ; assertValueSize ( serialized ) ; return [ [ keyEntries , { value : viewToArray ( serialized ) } ] ] ; } const entries = Object . entries ( keyEntries ) ; if ( entries . length > MAX_KEYS ) { throw new RangeError ( `Maximum number of pairs is ${ MAX_KEYS }.` ) ; } const result = [ ] ; for ( const [ key , rawValue ] of entries ) { assertKeySize ( key , true ) ; if ( rawValue === undefined ) continue ; const serialized = serialize ( rawValue ) ; assertValueSize ( serialized , key ) ; result . push ( [ key , { value : viewToArray ( serialized ) } ] ) ; } return result ; } function normaliseDeleteKeys ( keys ) { if ( Array . isArray ( keys ) ) { if ( keys . length > MAX_KEYS ) { throw new RangeError ( `Maximum number of keys is ${ MAX_KEYS }.` ) ; } const defined = [ ] ; for ( const key of keys ) { if ( key === undefined ) continue ; assertKeySize ( key , true ) ; defined . push ( key ) ; } return defined ; } else { assertKeySize ( keys ) ; return [ keys ] ; } } const kInner = Symbol ( "kInner" ) ; const kStartTxnCount = Symbol ( "kStartTxnCount" ) ; const kRolledback = Symbol ( "kRolledback" ) ; const kCommitted = Symbol ( "kCommitted" ) ; const kWriteSet = Symbol ( "kWriteSet" ) ; export class DurableObjectTransaction implements DurableObjectOperator { readonly [ kInner ] : ShadowStorage ; readonly [ kStartTxnCount ] : number ; [ kRolledback ] = false ; [ kCommitted ] = false ; readonly [ kWriteSet ] = new Set < string > ( ) ; constructor ( inner , startTxnCount ) { this [ kInner ] = new ShadowStorage ( inner ) ; this [ kStartTxnCount ] = startTxnCount ; } #check ( op ) { if ( this [ kRolledback ] ) { throw new Error ( `Cannot ${ op }() on rolled back transaction` ) ; } if ( this [ kCommitted ] ) { throw new Error ( `Cannot call ${ op }() on transaction that has already committed: did you move \`txn\` outside of the closure?` ) ; } } #markWritten ( ... keys ) { addAll ( this [ kWriteSet ] , keys ) ; if ( this [ kWriteSet ] . size > MAX_KEYS ) { throw new Error ( `Maximum number of keys modified in a transaction is ${ MAX_KEYS }.` ) ; } } get < Value = unknown > ( key , options ? ) ; get < Value = unknown > ( keys , options ? ) ; get < Value = unknown > ( keys , options ? ) { if ( keys === undefined ) { throw new TypeError ( "Failed to execute 'get' on 'DurableObjectTransaction'" + undefinedKeyError ) ; } this . #check ( "get" ) ; return runWithInputGateClosed ( ( ) => get ( this [ kInner ] , keys as any ) , options ?. allowConcurrency ) ; } #put < Value = unknown > ( keyEntries , valueOptions ? ) { const entries = normalisePutEntries ( keyEntries , valueOptions ) ; this . #markWritten ( ... entries . map ( ( [ key ] ) => key ) ) ; return this [ kInner ] . putMany ( entries ) ; } put < Value = unknown > ( key , value , options ? ) ; put < Value = unknown > ( entries , options ? ) ; put < Value = unknown > ( keyEntries , valueOptions ? , options ? ) { if ( keyEntries === undefined ) { throw new TypeError ( "Failed to execute 'put' on 'DurableObjectTransaction'" + undefinedKeyError ) ; } this . #check ( "put" ) ; if ( ! options && typeof keyEntries !== "string" ) options = valueOptions ; return waitUntilOnOutputGate ( runWithInputGateClosed ( ( ) => this . #put ( keyEntries , valueOptions ) , options ?. allowConcurrency ) , options ?. allowUnconfirmed ) ; } #delete ( keys ) { const keysIsArray = Array . isArray ( keys ) ; keys = normaliseDeleteKeys ( keys ) ; this . #markWritten ( ... keys ) ; return keysIsArray ? this [ kInner ] . deleteMany ( keys ) : Promise . resolve ( this [ kInner ] . delete ( keys [ 0 ] ) ) ; } delete ( key , options ? ) ; delete ( keys , options ? ) ; delete ( keys , options ? ) { if ( keys === undefined ) { throw new TypeError ( "Failed to execute 'delete' on 'DurableObjectTransaction'" + undefinedKeyError ) ; } this . #check ( "delete" ) ; return waitUntilOnOutputGate ( runWithInputGateClosed ( ( ) => this . #delete ( keys ) , options ?. allowConcurrency ) , options ?. allowUnconfirmed ) ; } deleteAll ( ) { throw new Error ( "Cannot call deleteAll() within a transaction" ) ; } list < Value = unknown > ( options = { } ) { this . #check ( "list" ) ; return runWithInputGateClosed ( ( ) => list ( this [ kInner ] , options ) , options . allowConcurrency ) ; } rollback ( ) { if ( this [ kRolledback ] ) return ; this . #check ( "rollback" ) ; this [ kRolledback ] = true ; } } const txnWriteSetsMaxSize = 16 ; function runWithGatesClosed < T > ( closure , options ? ) { return waitUntilOnOutputGate ( runWithInputGateClosed ( closure , options ?. allowConcurrency ) , options ?. allowUnconfirmed ) ; } export class DurableObjectStorage implements DurableObjectOperator { readonly #mutex = new ReadWriteMutex ( ) ; #txnCount = 0 ; readonly #txnWriteSets = new Map < number , Set < string > > ( ) ; #deletedKeySets : string [ ] [ ] = [ ] ; readonly #deletedKeyResults = new Map < string [ ] , number > ( ) ; readonly #inner : Storage ; readonly #shadow : ShadowStorage ; constructor ( inner ) { this . #inner = inner ; this . #shadow = new ShadowStorage ( inner , false ) ; } async #txnRead < T > ( closure ) { const startTxnCount = this . #txnCount ; const txn = new DurableObjectTransaction ( this . #shadow , startTxnCount ) ; const result = await closure ( txn ) ; txn [ kCommitted ] = true ; return { txn , result } ; } async #txnValidateWrite ( txn ) { if ( txn [ kRolledback ] ) return true ; return this . #mutex . runWithWrite ( async ( ) => { const finishTxnCount = this . #txnCount ; const readSet = txn [ kInner ] . readSet ! ; for ( let t = txn [ kStartTxnCount ] + 1 ; t <= finishTxnCount ; t ++ ) { const otherWriteSet = await this . #txnWriteSets . get ( t ) ; if ( ! otherWriteSet || intersects ( otherWriteSet , readSet ) ) { return false ; } } this . #txnRecordWriteSet ( txn [ kWriteSet ] ) ; for ( const [ key , value ] of txn [ kInner ] . copies . entries ( ) ) { this . #shadow . copies . set ( key , value ) ; } await this . #flush ( ) ; return true ; } ) ; } #txnRecordWriteSet ( writeSet ) { this . #txnCount ++ ; this . #txnWriteSets . set ( this . #txnCount , writeSet ) ; this . #txnWriteSets . delete ( this . #txnCount - txnWriteSetsMaxSize ) ; } transaction < T > ( closure ) { return runWithGatesClosed ( async ( ) => { while ( true ) { const { txn , result } = await this . #txnRead ( closure ) ; if ( await this . #txnValidateWrite ( txn ) ) return result ; } } ) ; } get < Value = unknown > ( key , options ? ) ; get < Value = unknown > ( keys , options ? ) ; async get < Value = unknown > ( keys , options ? ) { if ( keys === undefined ) { throw new TypeError ( "Failed to execute 'get' on 'DurableObjectStorage'" + undefinedKeyError ) ; } return runWithInputGateClosed ( ( ) => this . #mutex . runWithRead ( ( ) => get ( this . #shadow , keys as any ) ) , options ?. allowConcurrency ) ; } #flush = async ( ) : Promise < void > => { if ( this . #shadow . copies . size === 0 ) { assert . strictEqual ( this . #deletedKeySets . length , 0 ) ; return ; } const deletedKeySets = this . #deletedKeySets ; this . #deletedKeySets = [ ] ; const entries = [ ... this . #shadow . copies . entries ( ) ] ; const allDeletedKeys = new Set < string > ( ) ; for ( const deleteKeySet of deletedKeySets ) { const result = await this . #inner . deleteMany ( deleteKeySet ) ; this . #deletedKeyResults . set ( deleteKeySet , result ) ; addAll ( allDeletedKeys , deleteKeySet ) ; } const putEntries = [ ] ; const deleteKeys = [ ] ; for ( const [ key , value ] of entries ) { if ( value ) putEntries . push ( [ key , value ] ) ; else if ( ! allDeletedKeys . has ( key ) ) deleteKeys . push ( key ) ; } if ( putEntries . length > 0 ) await this . #inner . putMany ( putEntries ) ; if ( deleteKeys . length > 0 ) await this . #inner . deleteMany ( deleteKeys ) ; for ( const [ key , value ] of entries ) { if ( this . #shadow . copies . get ( key ) === value ) { this . #shadow . copies . delete ( key ) ; } } } ; put < Value = unknown > ( key , value , options ? ) ; put < Value = unknown > ( entries , options ? ) ; put < Value = unknown > ( keyEntries , valueOptions ? , options ? ) { if ( keyEntries === undefined ) { throw new TypeError ( "Failed to execute 'put' on 'DurableObjectStorage'" + undefinedKeyError ) ; } const entries = normalisePutEntries ( keyEntries , valueOptions ) ; if ( ! options && typeof keyEntries !== "string" ) options = valueOptions ; return runWithGatesClosed ( async ( ) => { await this . #mutex . runWithWrite ( ( ) => { for ( const [ key , value ] of entries ) this . #shadow . put ( key , value ) ; this . #txnRecordWriteSet ( new Set ( entries . map ( ( [ key ] ) => key ) ) ) ; } ) ; await Promise . resolve ( ) ; return this . #mutex . runWithWrite ( this . #flush ) ; } , options ) ; } delete ( key , options ? ) ; delete ( keys , options ? ) ; delete ( keys , options ? ) { if ( keys === undefined ) { throw new TypeError ( "Failed to execute 'delete' on 'DurableObjectStorage'" + undefinedKeyError ) ; } const keysIsArray = Array . isArray ( keys ) ; keys = normaliseDeleteKeys ( keys ) ; let deleted = 0 ; const deletedKeySet = [ ] ; return runWithGatesClosed ( async ( ) => { await this . #mutex . runWithWrite ( ( ) => { for ( const key of keys ) { if ( key === undefined ) continue ; if ( this . #shadow . copies . has ( key ) ) { if ( this . #shadow . copies . get ( key ) !== undefined ) { deleted ++ ; } } else { deletedKeySet . push ( key ) ; } this . #shadow . copies . set ( key , undefined ) ; } if ( deletedKeySet . length ) this . #deletedKeySets . push ( deletedKeySet ) ; this . #txnRecordWriteSet ( new Set ( keys ) ) ; } ) ; await Promise . resolve ( ) ; return this . #mutex . runWithWrite ( async ( ) => { await this . #flush ( ) ; if ( deletedKeySet . length ) { assert ( ! this . #deletedKeySets . includes ( deletedKeySet ) ) ; const result = this . #deletedKeyResults . get ( deletedKeySet ) ; this . #deletedKeyResults . delete ( deletedKeySet ) ; assert ( result !== undefined ) ; deleted += result ; } return keysIsArray ? deleted : deleted > 0 ; } ) ; } , options ) ; } async deleteAll ( options ? ) { return runWithGatesClosed ( ( ) => this . #mutex . runWithWrite ( async ( ) => { const { keys } = await this . #shadow . list ( ) ; const names = keys . map ( ( { name } ) => name ) ; for ( const key of names ) this . #shadow . copies . set ( key , undefined ) ; this . #txnRecordWriteSet ( new Set ( names ) ) ; await this . #flush ( ) ; } ) , options ) ; } async list < Value = unknown > ( options ? ) { return runWithInputGateClosed ( ( ) => this . #mutex . runWithRead ( ( ) => list ( this . #shadow , options ) ) , options ?. allowConcurrency ) ; } }