add docker mounting

Author: alexcos20

docs/API.md

@@ -1686,6 +1686,32 @@ List files in a bucket.
 
 ---
 
+### `HTTP` GET /api/services/persistentStorage/buckets/:bucketId/files/:fileName/object
+
+#### Description
+
+Return the `fileObject` for a specific file in a bucket (useful for passing references to other subsystems like compute).
+
+#### Query Parameters
+
+| name            | type   | required | description |
+| --------------- | ------ | -------- | ----------- |
+| consumerAddress | string | v        | consumer address |
+| signature       | string | v        | signed message (consumerAddress + nonce + command) |
+| nonce           | string | v        | request nonce |
+
+#### Response (200)
+
+```json
+{
+  "type": "nodePersistentStorage",
+  "bucketId": "uuid",
+  "fileName": "hello.txt"
+}
+```
+
+---
+
 ### `HTTP` POST /api/services/persistentStorage/buckets/:bucketId/files/:fileName
 
 #### Description

docs/persistentStorage.md

@@ -144,6 +144,7 @@ All persistent storage operations are implemented as protocol commands in the ha
 - `persistentStorageCreateBucket`
 - `persistentStorageGetBuckets`
 - `persistentStorageListFiles`
+- `persistentStorageGetFileObject`
 - `persistentStorageUploadFile`
 - `persistentStorageDeleteFile`
 
@@ -157,6 +158,7 @@ At a glance:
 - `POST /api/services/persistentStorage/buckets`
 - `GET /api/services/persistentStorage/buckets`
 - `GET /api/services/persistentStorage/buckets/:bucketId/files`
+- `GET /api/services/persistentStorage/buckets/:bucketId/files/:fileName/object`
 - `POST /api/services/persistentStorage/buckets/:bucketId/files/:fileName`
 - `DELETE /api/services/persistentStorage/buckets/:bucketId/files/:fileName`
 

src/@types/PersistentStorage.ts

@@ -1,5 +1,5 @@
 import type { AccessList } from './AccessList'
-
+import type { BaseFileObject } from './fileObject.js'
 export type PersistentStorageType = 'localfs' | 's3'
 
 export interface PersistentStorageLocalFSOptions {
@@ -22,3 +22,20 @@ export interface PersistentStorageConfig {
   accessLists: AccessList[]
   options: PersistentStorageLocalFSOptions | PersistentStorageS3Options
 }
+
+/**
+ * Docker mount descriptor used by the Docker C2D engine.
+ * Mirrors Dockerode `HostConfig.Mounts[]` item shape.
+ */
+export interface DockerMountObject {
+  Type: 'bind'
+  Source: string
+  Target: string
+  ReadOnly: boolean
+}
+
+export interface PersistentStorageObject extends BaseFileObject {
+  type: 'nodePersistentStorage'
+  bucketId: string
+  fileName: string
+}

src/@types/commands.ts

@@ -348,6 +348,14 @@ export interface PersistentStorageUploadFileCommand extends Command {
   fileName: string
 }
 
+export interface PersistentStorageGetFileObjectCommand extends Command {
+  consumerAddress: string
+  signature: string
+  nonce: string
+  bucketId: string
+  fileName: string
+}
+
 export interface PersistentStorageDeleteFileCommand extends Command {
   consumerAddress: string
   signature: string

src/components/c2d/compute_engine_docker.ts

@@ -1820,6 +1820,64 @@ export class C2DEngineDocker extends C2DEngine {
         }
         containerInfo.Env = envVars
       }
+      // persistent Storage: bind-mount bucket files into the job container (localfs backend)
+      for (const i in job.assets) {
+        const asset = job.assets[i]
+        if (!asset.fileObject || asset.fileObject.type !== 'nodePersistentStorage') {
+          continue
+        }
+        const fo = asset.fileObject as { bucketId?: string; fileName?: string }
+        if (!fo.bucketId || !fo.fileName) {
+          CORE_LOGGER.error(
+            `Job ${job.jobId} asset ${i}: nodePersistentStorage requires bucketId and fileName`
+          )
+          job.status = C2DStatusNumber.DataProvisioningFailed
+          job.statusText = C2DStatusText.DataProvisioningFailed
+          job.isRunning = false
+          job.dateFinished = String(Date.now() / 1000)
+          await this.db.updateJob(job)
+          await this.cleanupJob(job)
+          return
+        }
+        const ps = OceanNode.getInstance().getPersistentStorage()
+        if (!ps) {
+          CORE_LOGGER.error(
+            `Job ${job.jobId} asset ${i}: persistent storage is not configured on this node`
+          )
+          job.status = C2DStatusNumber.DataProvisioningFailed
+          job.statusText = C2DStatusText.DataProvisioningFailed
+          job.isRunning = false
+          job.dateFinished = String(Date.now() / 1000)
+          await this.db.updateJob(job)
+          await this.cleanupJob(job)
+          return
+        }
+        try {
+          const bindMount = await ps.getDockerMountObject(
+            fo.bucketId,
+            fo.fileName,
+            job.owner
+          )
+          CORE_LOGGER.debug(
+            `Mounting bucket ${fo.bucketId} to folder ${bindMount.Target}`
+          )
+          hostConfig.Mounts.push(bindMount)
+          mountVols[bindMount.Target] = {}
+        } catch (e) {
+          const errMsg = e instanceof Error ? e.message : String(e)
+          CORE_LOGGER.error(
+            `Job ${job.jobId} asset ${i}: failed to resolve persistent storage bind: ${errMsg}`
+          )
+          job.status = C2DStatusNumber.DataProvisioningFailed
+          job.statusText = C2DStatusText.DataProvisioningFailed
+          job.isRunning = false
+          job.dateFinished = String(Date.now() / 1000)
+          await this.db.updateJob(job)
+          await this.cleanupJob(job)
+          return
+        }
+      }
+
       const container = await this.createDockerContainer(containerInfo, true)
       if (container) {
         job.status = C2DStatusNumber.Provisioning
@@ -2767,6 +2825,10 @@ export class C2DEngineDocker extends C2DEngine {
       if (asset.fileObject) {
         try {
           if (asset.fileObject.type) {
+            if (asset.fileObject.type === 'nodePersistentStorage') {
+              // local storage is handled later, when we start the container and create the binds
+              continue
+            }
             storage = Storage.getStorageClass(asset.fileObject, config)
           } else {
             CORE_LOGGER.info('asset file object seems to be encrypted, checking it...')

src/components/core/compute/initialize.ts

@@ -39,6 +39,10 @@ import {
   validateAlgoForDataset,
   validateOutput
 } from './utils.js'
+import {
+  ensureConsumerAllowedForPersistentStorageLocalfsFileObject,
+  rejectPersistentStorageFileObjectOnAlgorithm
+} from '../../persistentStorage/PersistentStorageFactory.js'
 
 export class ComputeInitializeHandler extends CommandHandler {
   validate(command: ComputeInitializeCommand): ValidateParams {
@@ -220,6 +224,22 @@ export class ComputeInitializeHandler extends CommandHandler {
       if (isValidOutput.status.httpStatus !== 200) {
         return isValidOutput
       }
+      const algoPersistentStorageBan = rejectPersistentStorageFileObjectOnAlgorithm(
+        task.algorithm.fileObject
+      )
+      if (algoPersistentStorageBan) {
+        return algoPersistentStorageBan
+      }
+      for (const dataset of task.datasets) {
+        const psAccess = await ensureConsumerAllowedForPersistentStorageLocalfsFileObject(
+          node,
+          task.consumerAddress,
+          dataset.fileObject
+        )
+        if (psAccess) {
+          return psAccess
+        }
+      }
       // check algo
       let index = 0
       const policyServer = new PolicyServer()

src/components/core/compute/startCompute.ts

@@ -43,6 +43,10 @@ import { getNonceAsNumber } from '../utils/nonceHandler.js'
 import { PolicyServer } from '../../policyServer/index.js'
 import { checkCredentials } from '../../../utils/credentials.js'
 import { checkAddressOnAccessList } from '../../../utils/accessList.js'
+import {
+  ensureConsumerAllowedForPersistentStorageLocalfsFileObject,
+  rejectPersistentStorageFileObjectOnAlgorithm
+} from '../../persistentStorage/PersistentStorageFactory.js'
 
 export class CommonComputeHandler extends CommandHandler {
   validate(command: PaidComputeStartCommand): ValidateParams {
@@ -225,7 +229,23 @@ export class PaidComputeStartHandler extends CommonComputeHandler {
         }
       }
       const policyServer = new PolicyServer()
-      // check algo
+      const algoPersistentStorageBan = rejectPersistentStorageFileObjectOnAlgorithm(
+        task.algorithm.fileObject
+      )
+      if (algoPersistentStorageBan) {
+        return algoPersistentStorageBan
+      }
+      for (const dataset of task.datasets) {
+        const psAccess = await ensureConsumerAllowedForPersistentStorageLocalfsFileObject(
+          node,
+          task.consumerAddress,
+          dataset.fileObject
+        )
+        if (psAccess) {
+          return psAccess
+        }
+      }
+      // check algo and datasets (orders, credentials, etc.)
       for (const elem of [...[task.algorithm], ...task.datasets]) {
         const result: any = { validOrder: false }
         if ('documentId' in elem && elem.documentId) {
@@ -747,6 +767,22 @@ export class FreeComputeStartHandler extends CommonComputeHandler {
         return isValidOutput
       }
       const policyServer = new PolicyServer()
+      const algoPersistentStorageBanFree = rejectPersistentStorageFileObjectOnAlgorithm(
+        task.algorithm.fileObject
+      )
+      if (algoPersistentStorageBanFree) {
+        return algoPersistentStorageBanFree
+      }
+      for (const dataset of task.datasets) {
+        const psAccess = await ensureConsumerAllowedForPersistentStorageLocalfsFileObject(
+          thisNode,
+          task.consumerAddress,
+          dataset.fileObject
+        )
+        if (psAccess) {
+          return psAccess
+        }
+      }
       for (const elem of [...[task.algorithm], ...task.datasets]) {
         if (!('documentId' in elem)) {
           continue

src/components/core/handler/coreHandlersRegistry.ts

@@ -51,6 +51,7 @@ import {
   PersistentStorageCreateBucketHandler,
   PersistentStorageDeleteFileHandler,
   PersistentStorageGetBucketsHandler,
+  PersistentStorageGetFileObjectHandler,
   PersistentStorageListFilesHandler,
   PersistentStorageUploadFileHandler
 } from './persistentStorage.js'
@@ -190,6 +191,10 @@ export class CoreHandlersRegistry {
       PROTOCOL_COMMANDS.PERSISTENT_STORAGE_UPLOAD_FILE,
       new PersistentStorageUploadFileHandler(node)
     )
+    this.registerCoreHandler(
+      PROTOCOL_COMMANDS.PERSISTENT_STORAGE_GET_FILE_OBJECT,
+      new PersistentStorageGetFileObjectHandler(node)
+    )
     this.registerCoreHandler(
       PROTOCOL_COMMANDS.PERSISTENT_STORAGE_DELETE_FILE,
       new PersistentStorageDeleteFileHandler(node)

src/components/core/handler/persistentStorage.ts

@@ -3,6 +3,7 @@ import type {
   PersistentStorageCreateBucketCommand,
   PersistentStorageDeleteFileCommand,
   PersistentStorageGetBucketsCommand,
+  PersistentStorageGetFileObjectCommand,
   PersistentStorageListFilesCommand,
   PersistentStorageUploadFileCommand
 } from '../../../@types/commands.js'
@@ -220,6 +221,60 @@ export class PersistentStorageListFilesHandler extends CommandHandler {
   }
 }
 
+export class PersistentStorageGetFileObjectHandler extends CommandHandler {
+  validate(command: PersistentStorageGetFileObjectCommand): ValidateParams {
+    const base = validateCommandParameters(command, [
+      'consumerAddress',
+      'signature',
+      'nonce',
+      'bucketId',
+      'fileName'
+    ])
+    if (!base.valid) return base
+    return { valid: true }
+  }
+
+  async handle(task: PersistentStorageGetFileObjectCommand): Promise<P2PCommandResponse> {
+    const validationResponse = await this.verifyParamsAndRateLimits(task)
+    if (this.shouldDenyTaskHandling(validationResponse)) return validationResponse
+
+    const isAuthRequestValid = await this.validateTokenOrSignature(
+      task.authorization,
+      task.consumerAddress,
+      task.nonce,
+      task.signature,
+      task.command
+    )
+    if (isAuthRequestValid.status.httpStatus !== 200) return isAuthRequestValid
+
+    try {
+      const storage = await requirePersistentStorage(this)
+      const obj = await storage.getFileObject(
+        task.bucketId,
+        task.fileName,
+        task.consumerAddress
+      )
+      return {
+        stream: Readable.from(JSON.stringify(obj)),
+        status: { httpStatus: 200, error: null }
+      }
+    } catch (e) {
+      if (e instanceof PersistentStorageAccessDeniedError) {
+        return {
+          stream: null,
+          status: { httpStatus: 403, error: e.message }
+        }
+      }
+      const message = e instanceof Error ? e.message : String(e)
+      if (message.toLowerCase().includes('file not found')) {
+        return { stream: null, status: { httpStatus: 404, error: message } }
+      }
+      CORE_LOGGER.error(`PersistentStorageGetFileObjectHandler error: ${message}`)
+      return { stream: null, status: { httpStatus: 500, error: message } }
+    }
+  }
+}
+
 export class PersistentStorageUploadFileHandler extends CommandHandler {
   validate(command: PersistentStorageUploadFileCommand): ValidateParams {
     const base = validateCommandParameters(command, [

src/components/httpRoutes/persistentStorage.ts

@@ -9,6 +9,7 @@ import {
   PersistentStorageCreateBucketHandler,
   PersistentStorageDeleteFileHandler,
   PersistentStorageGetBucketsHandler,
+  PersistentStorageGetFileObjectHandler,
   PersistentStorageListFilesHandler,
   PersistentStorageUploadFileHandler
 } from '../core/handler/persistentStorage.js'
@@ -108,6 +109,36 @@ persistentStorageRoutes.get(
   }
 )
 
+// Get file object for a file in a bucket
+persistentStorageRoutes.get(
+  `${SERVICES_API_BASE_PATH}/persistentStorage/buckets/:bucketId/files/:fileName/object`,
+  async (req, res) => {
+    try {
+      const response = await new PersistentStorageGetFileObjectHandler(
+        req.oceanNode
+      ).handle({
+        command: PROTOCOL_COMMANDS.PERSISTENT_STORAGE_GET_FILE_OBJECT,
+        consumerAddress: req.query.consumerAddress as string,
+        signature: req.query.signature as string,
+        nonce: req.query.nonce as string,
+        bucketId: req.params.bucketId,
+        fileName: req.params.fileName,
+        authorization: req.headers?.authorization,
+        caller: req.caller
+      } as any)
+      if (!response.stream) {
+        res.status(response.status.httpStatus).send(response.status.error)
+        return
+      }
+      const payload = await streamToObject(response.stream as Readable)
+      res.status(200).json(payload)
+    } catch (error) {
+      HTTP_LOGGER.error(`PersistentStorage get file object error: ${error}`)
+      res.status(500).send('Internal Server Error')
+    }
+  }
+)
+
 // Upload file to bucket. Body is treated as raw bytes.
 persistentStorageRoutes.post(
   `${SERVICES_API_BASE_PATH}/persistentStorage/buckets/:bucketId/files/:fileName`,