refactor(internal): update native bridge and queue integration

- Improve native queue integration in protocol layer
- Update native bridge library loading
- Clean up queue native implementation
This commit is contained in:
Jeremie Fraeys 2026-02-18 12:45:59 -05:00
parent f9bba8a0fc
commit 96a8e139d5
No known key found for this signature in database
3 changed files with 31 additions and 19 deletions

View file

@ -248,22 +248,20 @@ func serializePacketToBuffer(p *ResponsePacket, buf []byte) ([]byte, error) {
return buf, nil
}
// appendString writes a string with varint length prefix
// appendString writes a string with fixed 16-bit length prefix
func appendString(buf []byte, s string) []byte {
length := uint64(len(s))
var tmp [binary.MaxVarintLen64]byte
n := binary.PutUvarint(tmp[:], length)
buf = append(buf, tmp[:n]...)
return append(buf, s...)
length := uint16(len(s))
buf = append(buf, byte(length>>8), byte(length))
buf = append(buf, s...)
return buf
}
// appendBytes writes bytes with varint length prefix
// appendBytes writes bytes with fixed 32-bit length prefix
func appendBytes(buf []byte, b []byte) []byte {
length := uint64(len(b))
var tmp [binary.MaxVarintLen64]byte
n := binary.PutUvarint(tmp[:], length)
buf = append(buf, tmp[:n]...)
return append(buf, b...)
length := uint32(len(b))
buf = append(buf, byte(length>>24), byte(length>>16), byte(length>>8), byte(length))
buf = append(buf, b...)
return buf
}
func appendUint32(buf []byte, value uint32) []byte {
@ -277,17 +275,17 @@ func (p *ResponsePacket) estimatedSize() int {
switch p.PacketType {
case PacketTypeSuccess:
return base + binary.MaxVarintLen64 + len(p.SuccessMessage)
return base + 2 + len(p.SuccessMessage)
case PacketTypeError:
return base + 1 + 2*binary.MaxVarintLen64 + len(p.ErrorMessage) + len(p.ErrorDetails)
return base + 1 + 2 + len(p.ErrorMessage) + 2 + len(p.ErrorDetails)
case PacketTypeProgress:
return base + 1 + 4 + 4 + binary.MaxVarintLen64 + len(p.ProgressMessage)
return base + 1 + 4 + 4 + 2 + len(p.ProgressMessage)
case PacketTypeStatus:
return base + binary.MaxVarintLen64 + len(p.StatusData)
return base + 2 + len(p.StatusData)
case PacketTypeData:
return base + binary.MaxVarintLen64 + len(p.DataType) + binary.MaxVarintLen64 + len(p.DataPayload)
return base + 2 + len(p.DataType) + 4 + len(p.DataPayload)
case PacketTypeLog:
return base + 1 + binary.MaxVarintLen64 + len(p.LogMessage)
return base + 1 + 2 + len(p.LogMessage)
default:
return base
}

View file

@ -13,6 +13,8 @@ import (
"os"
"time"
"unsafe"
"github.com/jfraeys/fetch_ml/internal/domain"
)
// UseNativeQueue controls whether to use C++ binary queue index.
@ -330,7 +332,7 @@ func (q *NativeQueue) RetryTask(task *Task) error {
return errors.New("failed to retry task")
}
RecordTaskRetry(task.JobName, ErrorUnknown)
RecordTaskRetry(task.JobName, domain.FailureUnknown)
return nil
}

View file

@ -11,6 +11,8 @@ import "C"
import (
"errors"
"unsafe"
"github.com/jfraeys/fetch_ml/internal/manifest"
)
// dirOverallSHA256HexNative implementation with native library.
@ -46,3 +48,13 @@ func GetSIMDImplName() string {
func HasSIMDSHA256() bool {
return C.fh_has_simd_sha256() == 1
}
// ScanArtifactsNative falls back to Go implementation.
func ScanArtifactsNative(runDir string) (*manifest.Artifacts, error) {
return ScanArtifacts(runDir)
}
// ExtractTarGzNative falls back to Go implementation.
func ExtractTarGzNative(archivePath, dstDir string) error {
return ExtractTarGz(archivePath, dstDir)
}