import { Dispatcher } from '../dispatcher' ; import { PinStore } from '../store/pin-store' ; import { ExecutionQueue } from './execution-queue' ; import { commitAtTime , ConflictResolution } from '../conflict-resolution' ; import { AnchorService , AnchorServiceResponse , AnchorStatus , CreateOpts , LoadOpts , InternalOpts , UpdateOpts , UnreachableCaseError , RunningStateLike , DiagnosticsLogger , StreamUtils , } from '@ceramicnetwork/common' ; import { RunningState } from './running-state' ; import CID from 'cids' ; import { catchError , concatMap , takeUntil } from 'rxjs/operators' ; import { empty , Observable , Subject , Subscription , timer } from 'rxjs' ; import { SnapshotState } from './snapshot-state' ; import { CommitID , StreamID } from '@ceramicnetwork/streamid' ; const APPLY_ANCHOR_COMMIT_ATTEMPTS = 3 ; export class StateManager { private private readonly syncedPinnedStreams : Set < string > = new Set ( ) ; param dispatcher param pinStore param executionQ param anchorService param conflictResolution param logger param fromMemoryOrStore param load constructor ( private readonly dispatcher , private readonly pinStore , private readonly executionQ , public anchorService , public conflictResolution , private readonly logger , private readonly fromMemoryOrStore , private readonly load ) { } param streamId wasPinnedStreamSynced ( streamId ) { return this . syncedPinnedStreams . has ( streamId . toString ( ) ) ; } param state$ param timeoutMillis async sync ( state$ , timeoutMillis ) { const tip$ = this . dispatcher . messageBus . queryNetwork ( state$ . id ) ; await tip$ . pipe ( takeUntil ( timer ( timeoutMillis ) ) , concatMap ( ( tip ) => this . _handleTip ( state$ , tip ) ) ) . toPromise ( ) ; if ( state$ . isPinned ) { this . syncedPinnedStreams . add ( state$ . id . toString ( ) ) ; } } param state$ async verifyLoneGenesis ( state$ ) { if ( state$ . value . log . length > 1 ) { return state$ ; } await this . conflictResolution . verifyLoneGenesis ( state$ . value ) ; return state$ ; } param state$ param commitId async atCommit ( state$ , commitId ) { return this . executionQ . forStream ( commitId ) . run ( async ( ) => { const snapshot = await this . conflictResolution . snapshotAtCommit ( state$ . value , commitId ) ; if ( StreamUtils . isStateSupersetOf ( snapshot , state$ . value ) ) { state$ . next ( snapshot ) ; } return new SnapshotState ( snapshot ) ; } ) ; } param state$ param timestamp atTime ( state$ , timestamp ) { const commitId = commitAtTime ( state$ , timestamp ) ; return this . atCommit ( state$ , commitId ) ; } param state$ param opts private applyWriteOpts ( state$ , opts ) { const anchor = ( opts as any ) . anchor ; const publish = ( opts as any ) . publish ; if ( anchor ) { this . anchor ( state$ ) ; } if ( publish ) { this . publishTip ( state$ ) ; } } param state$ param cid param opts returns private private async _handleTip ( state$ , cid , opts = { } ) { opts . throwOnInvalidCommit = opts . throwOnInvalidCommit ?? false ; this . logger . verbose ( `Learned of new tip ${ cid . toString ( ) } for stream ${ state$ . id . toString ( ) }` ) ; const next = await this . conflictResolution . applyTip ( state$ . value , cid , opts ) ; if ( next ) { state$ . next ( next ) ; this . logger . verbose ( `Stream ${ state$ . id . toString ( ) } successfully updated to tip ${ cid . toString ( ) }` ) ; await this . _updateStateIfPinned ( state$ ) ; return true ; } else { return false ; } } private async _updateStateIfPinned ( state$ ) { const isPinned = Boolean ( await this . pinStore . stateStore . load ( state$ . id ) ) ; if ( isPinned ) { await this . pinStore . add ( state$ ) ; } } publishTip ( state$ ) { this . dispatcher . publishTip ( state$ . id , state$ . tip ) ; } param streamId param tip private update ( streamId , tip ) { this . executionQ . forStream ( streamId ) . add ( async ( ) => { const state$ = await this . fromMemoryOrStore ( streamId ) ; if ( state$ ) await this . _handleTip ( state$ , tip ) ; } ) ; } param streamId param commit param opts async applyCommit ( streamId , commit , opts ) { return this . executionQ . forStream ( streamId ) . run ( async ( ) => { const state$ = await this . load ( streamId , opts ) ; const cid = await this . dispatcher . storeCommit ( commit , streamId ) ; await this . _handleTip ( state$ , cid , opts ) ; return state$ ; } ) ; } param state$ param tip param anchorCommit private private async _handleAnchorCommit ( state$ , tip , anchorCommit ) { for ( let remainingRetries = APPLY_ANCHOR_COMMIT_ATTEMPTS - 1 ; remainingRetries >= 0 ; remainingRetries -- ) { try { await this . executionQ . forStream ( state$ . id ) . run ( async ( ) => { const applied = await this . _handleTip ( state$ , anchorCommit ) ; if ( applied ) { this . publishTip ( state$ ) ; if ( remainingRetries < APPLY_ANCHOR_COMMIT_ATTEMPTS - 1 ) { this . logger . imp ( `Successfully applied anchor commit ${ anchorCommit . toString ( ) } for stream ${ state$ . id . toString ( ) }` ) ; } } } ) ; return ; } catch ( error ) { this . logger . warn ( `Error while applying anchor commit ${ anchorCommit . toString ( ) } for stream ${ state$ . id . toString ( ) }, ${ remainingRetries } retries remain. ${ error }` ) ; if ( remainingRetries == 0 ) { this . logger . err ( `Anchor failed for commit ${ tip . toString ( ) } of stream ${ state$ . id . toString ( ) }: ${ error }` ) ; if ( tip . equals ( state$ . tip ) ) { state$ . next ( { ... state$ . value , anchorStatus : AnchorStatus . FAILED } ) ; } } } } } /**
     * Request anchor for the latest stream state
     */ anchor ( state$ ) { if ( ! this . anchorService ) { throw new Error ( `Anchor requested for stream ${ state$ . id . toString ( ) } but anchoring is disabled` ) ; } if ( state$ . value . anchorStatus == AnchorStatus . ANCHORED ) { return Subscription . EMPTY ; } const anchorStatus$ = this . anchorService . requestAnchor ( state$ . id , state$ . tip ) ; return this . _processAnchorResponse ( state$ , anchorStatus$ ) ; } /**
     * Restart polling and handle response for a previously submitted anchor request
     */ confirmAnchorResponse ( state$ ) { const anchorStatus$ = this . anchorService . pollForAnchorResponse ( state$ . id , state$ . tip ) ; return this . _processAnchorResponse ( state$ , anchorStatus$ ) ; } private _processAnchorResponse ( state$ , anchorStatus$ ) { const stopSignal = new Subject < void > ( ) ; const subscription = anchorStatus$ . pipe ( takeUntil ( stopSignal ) , concatMap ( async ( asr ) => { if ( ! asr . cid . equals ( state$ . tip ) && asr . status != AnchorStatus . ANCHORED ) { return ; } switch ( asr . status ) { case AnchorStatus . PENDING : { const next = { ... state$ . value , anchorStatus : AnchorStatus . PENDING , } ; if ( asr . anchorScheduledFor ) next . anchorScheduledFor = asr . anchorScheduledFor ; state$ . next ( next ) ; await this . _updateStateIfPinned ( state$ ) ; return ; } case AnchorStatus . PROCESSING : { state$ . next ( { ... state$ . value , anchorStatus : AnchorStatus . PROCESSING } ) ; await this . _updateStateIfPinned ( state$ ) ; return ; } case AnchorStatus . ANCHORED : { await this . _handleAnchorCommit ( state$ , asr . cid , asr . anchorCommit ) ; stopSignal . next ( ) ; return ; } case AnchorStatus . FAILED : { this . logger . warn ( `Anchor failed for commit ${ asr . cid . toString ( ) } of stream ${ asr . streamId }: ${ asr . message }` ) ; state$ . next ( { ... state$ . value , anchorStatus : AnchorStatus . FAILED } ) ; stopSignal . next ( ) ; return ; } default : throw new UnreachableCaseError ( asr , 'Unknown anchoring state' ) ; } } ) , catchError ( ( error ) => { this . logger . warn ( `Error while anchoring stream ${ state$ . id . toString ( ) }:${ error }` ) ; this . logger . warn ( error ) ; return empty ( ) ; } ) ) . subscribe ( ) ; state$ . add ( subscription ) ; return subscription ; } }