storage/splitfdstream: add lookup_digest RPC method

Add a new lookup_digest JSON-RPC method that allows retrieving a file or
chunk by its digest.  This enables efficient content-addressed lookup
using the chunked package's cache infrastructure.

Signed-off-by: Giuseppe Scrivano <gscrivan@redhat.com>

Author: giuseppe

storage/pkg/splitfdstream/protocol.go

@@ -14,6 +14,7 @@ const (
 const (
 	MethodApplySplitFDStream = "apply_splitfdstream"
 	MethodGetSplitFDStream   = "get_splitfdstream"
+	MethodLookupDigest       = "lookup_digest"
 	MethodPing               = "ping"
 )
 
@@ -28,6 +29,8 @@ const (
 	// Application-specific error codes (range -32000 to -32099)
 	ErrorCodeDriverNotSupported  = -32000
 	ErrorCodeFileDescriptorError = -32001
+	ErrorCodeDigestNotFound      = -32002
+	ErrorCodeStoreNotAvailable   = -32003
 )
 
 // SplitFDStreamRequest represents a JSON-RPC request for splitfdstream operations.
@@ -59,6 +62,8 @@ type SplitFDStreamResult struct {
 	FileDescriptors *int   `json:"fileDescriptors,omitempty"`
 	BatchSize       int    `json:"batchSize,omitempty"`
 	NumBatches      int    `json:"numBatches,omitempty"`
+	Offset          int64  `json:"offset,omitempty"`
+	ChunkSize       int64  `json:"chunkSize,omitempty"`
 	Message         string `json:"message,omitempty"`
 }
 
@@ -167,6 +172,8 @@ func (r *SplitFDStreamRequest) IsValid() error {
 				Data:    "layerId is required for get_splitfdstream",
 			}
 		}
+	case MethodLookupDigest:
+		// Validation is done in ParseLookupDigestParams
 	case MethodPing:
 		// No validation required for ping
 	default:
@@ -270,3 +277,37 @@ func ParsePingParams(params SplitFDStreamParams) (*PingParams, error) {
 
 	return &pingParams, nil
 }
+
+// LookupDigestParams represents parameters specific to lookup_digest method.
+type LookupDigestParams struct {
+	Digest string `json:"digest"`
+}
+
+// ParseLookupDigestParams parses the generic params as LookupDigestParams.
+func ParseLookupDigestParams(params SplitFDStreamParams) (*LookupDigestParams, error) {
+	// Merge Options into a flat map
+	flatMap := make(map[string]interface{})
+	for k, v := range params.Options {
+		flatMap[k] = v
+	}
+
+	data, err := json.Marshal(flatMap)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal params: %w", err)
+	}
+
+	var lookupParams LookupDigestParams
+	if err := json.Unmarshal(data, &lookupParams); err != nil {
+		return nil, fmt.Errorf("failed to parse lookup params: %w", err)
+	}
+
+	if lookupParams.Digest == "" {
+		return nil, &SplitFDStreamError{
+			Code:    ErrorCodeInvalidParams,
+			Message: "Invalid params",
+			Data:    "digest is required for lookup_digest",
+		}
+	}
+
+	return &lookupParams, nil
+}

storage/pkg/splitfdstream/server.go

@@ -70,22 +70,30 @@ func (r *limitedSocketReader) Read(p []byte) (int, error) {
 	return n, nil
 }
 
+// DigestLookupFunc is a function that looks up a file by its digest.
+// It returns the open file, offset within the file, and chunk size.
+// A nil file means the digest was not found.
+// The caller is responsible for closing the returned file.
+type DigestLookupFunc func(digest string) (file *os.File, offset int64, size int64, err error)
+
 // JSONRPCServer manages the JSON-RPC server for storage operations.
 type JSONRPCServer struct {
-	driver      any
-	listener    net.Listener
-	socketPath  string
-	shutdown    chan struct{}
-	connections sync.WaitGroup
-	mu          sync.RWMutex
-	running     bool
+	driver       any
+	digestLookup DigestLookupFunc
+	listener     net.Listener
+	socketPath   string
+	shutdown     chan struct{}
+	connections  sync.WaitGroup
+	mu           sync.RWMutex
+	running      bool
 }
 
 // NewJSONRPCServer creates a new JSON-RPC server.
-func NewJSONRPCServer(driver any) *JSONRPCServer {
+func NewJSONRPCServer(driver any, digestLookup DigestLookupFunc) *JSONRPCServer {
 	return &JSONRPCServer{
-		driver:   driver,
-		shutdown: make(chan struct{}),
+		driver:       driver,
+		digestLookup: digestLookup,
+		shutdown:     make(chan struct{}),
 	}
 }
 
@@ -271,6 +279,8 @@ func (s *JSONRPCServer) handleRequest(fdPasser *FDPasser, requestLine string) {
 		s.handleApplySplitFDStream(fdPasser, req)
 	case MethodGetSplitFDStream:
 		s.handleGetSplitFDStream(fdPasser, req)
+	case MethodLookupDigest:
+		s.handleLookupDigest(fdPasser, req)
 	default:
 		resp := NewErrorResponse(ErrorCodeMethodNotFound, "Method not found", req.Method, req.ID)
 		s.sendResponse(fdPasser, resp)
@@ -528,6 +538,71 @@ func (s *JSONRPCServer) handleGetSplitFDStream(fdPasser *FDPasser, req *SplitFDS
 	}
 }
 
+// handleLookupDigest handles lookup_digest requests.
+// Protocol:
+// 1. Client sends JSON-RPC request with digest
+// 2. Server looks up the digest in the cache
+// 3. Server sends response with offset and size, plus the file descriptor
+func (s *JSONRPCServer) handleLookupDigest(fdPasser *FDPasser, req *SplitFDStreamRequest) {
+	// Parse the lookup-specific parameters
+	lookupParams, err := ParseLookupDigestParams(req.Params)
+	if err != nil {
+		resp := NewErrorResponse(ErrorCodeInvalidParams, err.Error(), nil, req.ID)
+		s.sendResponse(fdPasser, resp)
+		return
+	}
+
+	// Check if digest lookup is available
+	if s.digestLookup == nil {
+		resp := NewErrorResponse(
+			ErrorCodeStoreNotAvailable,
+			"digest lookup is not available",
+			nil,
+			req.ID,
+		)
+		s.sendResponse(fdPasser, resp)
+		return
+	}
+
+	// Look up the digest
+	file, offset, size, err := s.digestLookup(lookupParams.Digest)
+	if err != nil {
+		resp := NewErrorResponse(ErrorCodeInternalError, fmt.Sprintf("failed to lookup digest: %v", err), nil, req.ID)
+		s.sendResponse(fdPasser, resp)
+		return
+	}
+
+	if file == nil {
+		resp := NewErrorResponse(
+			ErrorCodeDigestNotFound,
+			fmt.Sprintf("digest %s not found", lookupParams.Digest),
+			nil,
+			req.ID,
+		)
+		s.sendResponse(fdPasser, resp)
+		return
+	}
+
+	defer file.Close()
+
+	// Send success response with metadata
+	responseResult := &SplitFDStreamResult{
+		Offset:    offset,
+		ChunkSize: size,
+		Message:   "found",
+	}
+	numFDs := 1
+	responseResult.FileDescriptors = &numFDs
+	resp := NewSuccessResponse(responseResult, req.ID)
+	s.sendResponse(fdPasser, resp)
+
+	// Send the file descriptor
+	if err := fdPasser.SendFileDescriptors([]*os.File{file}, []byte{0}); err != nil {
+		logrus.Errorf("splitfdstream: failed to send file descriptor for digest lookup: %v", err)
+		return
+	}
+}
+
 // waitForContinue reads "continue\n" from the client.
 func (s *JSONRPCServer) waitForContinue(fdPasser *FDPasser) error {
 	msg, err := fdPasser.ReadLine()

storage/pkg/splitfdstream/server_test.go

@@ -31,7 +31,7 @@ func (m *MockDriver) GetSplitFDStream(id, parent string, options *GetSplitFDStre
 
 func TestSplitFDStreamServer_StartStop(t *testing.T) {
 	mockDriver := &MockDriver{}
-	server := NewJSONRPCServer(mockDriver)
+	server := NewJSONRPCServer(mockDriver, nil)
 
 	tmpDir := t.TempDir()
 	socketPath := filepath.Join(tmpDir, "test.sock")
@@ -63,7 +63,7 @@ func TestSplitFDStreamServer_StartStop(t *testing.T) {
 
 func TestSplitFDStreamServer_DoubleStart(t *testing.T) {
 	mockDriver := &MockDriver{}
-	server := NewJSONRPCServer(mockDriver)
+	server := NewJSONRPCServer(mockDriver, nil)
 
 	tmpDir := t.TempDir()
 	socketPath := filepath.Join(tmpDir, "test.sock")
@@ -84,7 +84,7 @@ func TestSplitFDStreamServer_DoubleStart(t *testing.T) {
 
 func TestSplitFDStreamServer_PingRequest(t *testing.T) {
 	mockDriver := &MockDriver{}
-	server := NewJSONRPCServer(mockDriver)
+	server := NewJSONRPCServer(mockDriver, nil)
 
 	tmpDir := t.TempDir()
 	socketPath := filepath.Join(tmpDir, "test.sock")
@@ -128,7 +128,7 @@ func TestSplitFDStreamServer_PingRequest(t *testing.T) {
 func TestSplitFDStreamServer_UnsupportedDriver(t *testing.T) {
 	// Use mockBaseDriver which does NOT implement SplitFDStreamDriver
 	baseDriver := &mockBaseDriver{}
-	server := NewJSONRPCServer(baseDriver)
+	server := NewJSONRPCServer(baseDriver, nil)
 
 	tmpDir := t.TempDir()
 	socketPath := filepath.Join(tmpDir, "test.sock")
@@ -182,7 +182,7 @@ func TestSplitFDStreamServer_UnsupportedDriver(t *testing.T) {
 
 func TestSplitFDStreamServer_InvalidRequest(t *testing.T) {
 	mockDriver := &MockDriver{}
-	server := NewJSONRPCServer(mockDriver)
+	server := NewJSONRPCServer(mockDriver, nil)
 
 	tmpDir := t.TempDir()
 	socketPath := filepath.Join(tmpDir, "test.sock")

storage/pkg/splitfdstream/server_unsupported.go

@@ -7,11 +7,15 @@ import (
 	"os"
 )
 
+// DigestLookupFunc is a function that looks up a file by its digest.
+// A nil file means the digest was not found.
+type DigestLookupFunc func(digest string) (file *os.File, offset int64, size int64, err error)
+
 // JSONRPCServer manages the JSON-RPC server for storage operations.
 type JSONRPCServer struct{}
 
 // NewJSONRPCServer creates a new JSON-RPC server.
-func NewJSONRPCServer(driver any) *JSONRPCServer {
+func NewJSONRPCServer(driver any, digestLookup DigestLookupFunc) *JSONRPCServer {
 	return &JSONRPCServer{}
 }
 

storage/store.go

@@ -4139,7 +4139,7 @@ func (s *store) SplitFDStreamSocket() (*os.File, error) {
 
 	// Initialize server if not already created
 	if s.jsonRPCServer == nil {
-		s.jsonRPCServer = splitfdstream.NewJSONRPCServer(s.graphDriver)
+		s.jsonRPCServer = splitfdstream.NewJSONRPCServer(s.graphDriver, nil)
 	}
 
 	// Start handling the server connection in a goroutine