移除 runtime 的 Processor v2 兼容层

packages/runtime/src/full-service.ts

@@ -1,51 +1,16 @@
 import { create } from '@bufbuild/protobuf'
 import { type HandlerContext, type ServiceImpl } from '@connectrpc/connect'
 import {
-  type Empty,
   ExecutionConfigSchema,
-  type PreprocessStreamRequest,
-  type ProcessBindingsRequest,
   type ProcessConfigRequest,
-  Processor,
   ProcessorV3,
   type ProcessStreamRequest,
   type StartRequest,
   type UpdateTemplatesRequest
 } from '@sentio/protos'
-import { ProcessorServiceImpl } from './service.js'
 import { ProcessorServiceImplV3 } from './service-v3.js'
 import { GLOBAL_CONFIG } from './global-config.js'
 
-export class FullProcessorServiceImpl implements ServiceImpl<typeof Processor> {
-  constructor(readonly instance: ProcessorServiceImpl) {}
-
-  async getConfig(request: ProcessConfigRequest, context: HandlerContext) {
-    const config = await this.instance.getConfig(request, context)
-    config.executionConfig = create(ExecutionConfigSchema, GLOBAL_CONFIG.execution)
-    return config
-  }
-
-  async start(request: StartRequest, context: HandlerContext) {
-    return await this.instance.start(request, context)
-  }
-
-  async stop(request: Empty, context: HandlerContext) {
-    return await this.instance.stop(request, context)
-  }
-
-  async processBindings(request: ProcessBindingsRequest, context: HandlerContext) {
-    return await this.instance.processBindings(request, context)
-  }
-
-  async *processBindingsStream(requests: AsyncIterable<ProcessStreamRequest>, context: HandlerContext) {
-    yield* this.instance.processBindingsStream(requests, context)
-  }
-
-  async *preprocessBindingsStream(requests: AsyncIterable<PreprocessStreamRequest>, context: HandlerContext) {
-    yield* this.instance.preprocessBindingsStream(requests, context)
-  }
-}
-
 export class FullProcessorServiceV3Impl implements ServiceImpl<typeof ProcessorV3> {
   constructor(readonly instance: ProcessorServiceImplV3) {}
 

packages/runtime/src/index.ts

@@ -3,7 +3,7 @@ export * from './state.js'
 export * from './utils.js'
 export * from './endpoints.js'
 export * from './chain-config.js'
-export * from './service.js'
+export * from './service-v3.js'
 export { GLOBAL_CONFIG, type GlobalConfig } from './global-config.js'
 export * from './db-context.js'
 export * from './provider.js'

packages/runtime/src/processor-runner.ts

@@ -10,14 +10,13 @@ import { Session } from 'node:inspector/promises'
 import { fork, ChildProcess } from 'child_process'
 import { fileURLToPath } from 'url'
 
-import { ProcessorServiceImpl } from './service.js'
 import { configureEndpoints } from './endpoints.js'
-import { FullProcessorServiceImpl, FullProcessorServiceV3Impl } from './full-service.js'
+import { FullProcessorServiceV3Impl } from './full-service.js'
 import { setupLogger } from './logger.js'
 
 import { setupOTLP } from './otlp.js'
 import { ActionServer } from './action-server.js'
-import { Processor, ProcessorV3 } from '@sentio/protos'
+import { ProcessorV3 } from '@sentio/protos'
 import { ProcessorServiceImplV3 } from './service-v3.js'
 import { dirname, join } from 'path'
 import { program, ProcessorRuntimeOptions } from './processor-runner-program.js'
@@ -67,7 +66,6 @@ if (!isChildProcess) {
 
 let server: any
 let processorHttp2Server: http2.Http2Server | undefined
-let baseService: ProcessorServiceImpl
 let httpServer: http.Server | undefined
 
 const loader = async () => {
@@ -82,15 +80,9 @@ if (options.startActionServer) {
 } else {
   const shutdown = () => processorHttp2Server?.close()
 
-  // for V2
-  baseService = new ProcessorServiceImpl(loader, options, shutdown)
-  const serviceV2 = new FullProcessorServiceImpl(baseService)
-
-  // for V3
   const serviceV3 = new FullProcessorServiceV3Impl(new ProcessorServiceImplV3(loader, options, shutdown))
 
   const routes = (router: ConnectRouter) => {
-    router.service(Processor, serviceV2)
     router.service(ProcessorV3, serviceV3)
   }
 
@@ -208,9 +200,6 @@ process
   })
   .on('uncaughtException', (err) => {
     console.error('Uncaught Exception, please checking if await is properly used', err)
-    if (baseService) {
-      baseService.unhandled = err
-    }
     // shutdownServers(1)
   })
   .on('unhandledRejection', (reason, _p) => {
@@ -219,9 +208,6 @@ process
       return
     }
     console.error('Unhandled Rejection, please checking if await is properly', reason)
-    if (baseService) {
-      baseService.unhandled = reason as Error
-    }
     // shutdownServers(1)
   })
 

packages/runtime/src/service-v3.ts

@@ -17,10 +17,9 @@ import { PluginManager } from './plugin.js'
 import { Subject } from 'rxjs'
 import { from } from 'ix/asynciterable'
 import { withAbort } from 'ix/asynciterable/operators'
-import { errorString } from './utils.js'
+import { errorString, recordRuntimeInfo } from './utils.js'
 
 import { processMetrics } from './metrics.js'
-import { recordRuntimeInfo } from './service.js'
 import { DataBindingContext } from './db-context.js'
 import { freezeGlobalConfig } from './global-config.js'
 import { ProcessorRuntimeOptions } from './processor-runner-program.js'
@@ -168,6 +167,8 @@ export class ProcessorServiceImplV3 implements ServiceImpl<typeof ProcessorV3> {
     PluginManager.INSTANCE.processBinding(binding, undefined, context)
       .then(async (result) => {
         await context.awaitPendings()
+        recordRuntimeInfo(result, binding.handlerType)
+
         const timeseriesResult = result.timeseriesResult
         for (let i = 0; i < timeseriesResult.length; i += TIME_SERIES_RESULT_BATCH_SIZE) {
           const batch = timeseriesResult.slice(i, i + TIME_SERIES_RESULT_BATCH_SIZE)
@@ -188,7 +189,6 @@ export class ProcessorServiceImplV3 implements ServiceImpl<typeof ProcessorV3> {
             value: WRITE_V2_EVENT_LOGS ? otherResults : create(ProcessResultSchema, { states: result.states })
           }
         })
-        recordRuntimeInfo(result, binding.handlerType)
       })
       .catch((e) => {
         console.error(e, e.stack)