logos-messaging/sds-go-bindings
1
0
Fork 0
mirror of https://github.com/logos-messaging/sds-go-bindings.git synced 2026-07-23 16:33:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: adapt bindings to nim-sds v0.4 JSON sds_* ABI

Browse Source 
nim-sds release/v0.4 moved to nim-ffi v0.1.5 and now exposes a
JSON + snake_case `sds_*` C ABI (sds_create, sds_wrap_outgoing_message,
sds_unwrap_received_message, ...). Rewrite the cgo layer to match:
requests/responses are marshalled as JSON, with seq[byte] carried as a
JSON number array (jsonByteArray) and retrievalHint as base64.

Also add repair_ready event support (OnRepairReady) and guard the
manager registry with rmRegistryMu: libsds fires the global callbacks on
its own worker threads, which raced the register/unregister writes done
on the manager-owning goroutines.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Ivan FB 2026-06-19 09:54:36 +02:00
parent 401a7671f0
commit 9ce92dedb2
No known key found for this signature in database
GPG Key ID: DF0C67A04C543270
2 changed files with 288 additions and 242 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

486
sds/sds.go
View File

@ -4,8 +4,8 @@ package sds
/*
#include <libsds.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
extern void sdsGlobalEventCallback(int ret, char* msg, size_t len, void* userData);
@ -15,21 +15,36 @@ package sds
int ret;
char* msg;
size_t len;
void* ffiWg;
} SdsResp;
static void* allocResp(void* wg) {
SdsResp* r = calloc(1, sizeof(SdsResp));
r->ffiWg = wg;
return r;
static void* allocResp(void) {
return calloc(1, sizeof(SdsResp));
}
static void freeResp(void* resp) {
if (resp != NULL) {
free(resp);
SdsResp* r = (SdsResp*) resp;
if (r->msg != NULL) {
free(r->msg);
}
free(r);
}
}
// libsds frees the buffer it hands the callback as soon as the callback
// returns, so the Go side must copy it before the waiting goroutine reads
// it. cGoMemDup makes that copy onto the C heap (freed by freeResp).
static char* cGoMemDup(const void* src, size_t len) {
if (src == NULL || len == 0) {
return NULL;
}
char* dst = (char*) malloc(len);
if (dst != NULL) {
memcpy(dst, src, len);
}
return dst;
}
static char* getMyCharPtr(void* resp) {
if (resp == NULL) {
return NULL;
@ -54,96 +69,57 @@ package sds
return m->ret;
}
// resp must be set != NULL in case interest on retrieving data from the callback
// SdsGoCallback is the result callback for the request/response FFI calls.
void SdsGoCallback(int ret, char* msg, size_t len, void* resp);
static void* cGoSdsNewReliabilityManager(void* resp) {
// We pass NULL because we are not interested in retrieving data from this callback
void* ret = SdsNewReliabilityManager((SdsCallBack) SdsGoCallback, resp);
return ret;
static void* cGoSdsCreate(const char* configJson, void* resp) {
return sds_create(configJson, (SdsCallBack) SdsGoCallback, resp);
}
static void cGoSdsSetEventCallback(void* rmCtx) {
// The 'sdsGlobalEventCallback' Go function is shared amongst all possible Reliability Manager instances.
// Given that the 'sdsGlobalEventCallback' is shared, we pass again the
// rmCtx instance but in this case is needed to pick up the correct method
// that will handle the event.
// In other words, for every call libsds makes to sdsGlobalEventCallback,
// the 'userData' parameter will bring the context of the rm that registered
// that sdsGlobalEventCallback.
// This technique is needed because cgo only allows to export Go functions and not methods.
SdsSetEventCallback(rmCtx, (SdsCallBack) sdsGlobalEventCallback, rmCtx);
static void cGoSdsSetEventCallback(void* ctx) {
// 'sdsGlobalEventCallback' is shared by all manager instances; we pass the
// ctx as userData so the dispatcher can route the event to the instance
// that registered it (cgo can export Go funcs but not methods).
sds_set_event_callback(ctx, (SdsCallBack) sdsGlobalEventCallback, ctx);
}
static void cGoSdsSetRetrievalHintProvider(void* rmCtx) {
SdsSetRetrievalHintProvider(rmCtx, (SdsRetrievalHintProvider) sdsGlobalRetrievalHintProvider, rmCtx);
static int cGoSdsSetRetrievalHintProvider(void* ctx) {
return sds_set_retrieval_hint_provider(ctx, (SdsRetrievalHintProvider) sdsGlobalRetrievalHintProvider, ctx);
}
static void cGoSdsCleanupReliabilityManager(void* rmCtx, void* resp) {
SdsCleanupReliabilityManager(rmCtx, (SdsCallBack) SdsGoCallback, resp);
static int cGoSdsWrapOutgoingMessage(void* ctx, const char* reqJson, void* resp) {
return sds_wrap_outgoing_message(ctx, (SdsCallBack) SdsGoCallback, resp, reqJson);
}
static void cGoSdsResetReliabilityManager(void* rmCtx, void* resp) {
SdsResetReliabilityManager(rmCtx, (SdsCallBack) SdsGoCallback, resp);
static int cGoSdsUnwrapReceivedMessage(void* ctx, const char* reqJson, void* resp) {
return sds_unwrap_received_message(ctx, (SdsCallBack) SdsGoCallback, resp, reqJson);
}
static void cGoSdsWrapOutgoingMessage(void* rmCtx,
void* message,
size_t messageLen,
const char* messageId,
const char* channelId,
void* resp) {
SdsWrapOutgoingMessage(rmCtx,
message,
messageLen,
messageId,
channelId,
(SdsCallBack) SdsGoCallback,
resp);
}
static void cGoSdsUnwrapReceivedMessage(void* rmCtx,
void* message,
size_t messageLen,
void* resp) {
SdsUnwrapReceivedMessage(rmCtx,
message,
messageLen,
(SdsCallBack) SdsGoCallback,
resp);
static int cGoSdsMarkDependenciesMet(void* ctx, const char* reqJson, void* resp) {
return sds_mark_dependencies_met(ctx, (SdsCallBack) SdsGoCallback, resp, reqJson);
}
static void cGoSdsMarkDependenciesMet(void* rmCtx,
char** messageIDs,
size_t count,
const char* channelId,
void* resp) {
SdsMarkDependenciesMet(rmCtx,
messageIDs,
count,
channelId,
(SdsCallBack) SdsGoCallback,
resp);
static int cGoSdsReset(void* ctx, void* resp) {
return sds_reset(ctx, (SdsCallBack) SdsGoCallback, resp);
}
static void cGoSdsStartPeriodicTasks(void* rmCtx, void* resp) {
SdsStartPeriodicTasks(rmCtx, (SdsCallBack) SdsGoCallback, resp);
static int cGoSdsStartPeriodicTasks(void* ctx, void* resp) {
return sds_start_periodic_tasks(ctx, (SdsCallBack) SdsGoCallback, resp);
}
static int cGoSdsDestroy(void* ctx, void* resp) {
return sds_destroy(ctx, (SdsCallBack) SdsGoCallback, resp);
}
*/
import "C"
import (
"encoding/json"
"errors"
"fmt"
"strconv"
"strings"
"sync"
"unsafe"
errorspkg "github.com/pkg/errors"
"go.uber.org/zap"
)
@ -151,16 +127,109 @@ var (
errEmptyReliabilityManager = errors.New("empty reliability manager")
)
//export SdsGoCallback
func SdsGoCallback(ret C.int, msg *C.char, len C.size_t, resp unsafe.Pointer) {
if resp != nil {
m := (*C.SdsResp)(resp)
m.ret = ret
m.msg = msg
m.len = len
wg := (*sync.WaitGroup)(m.ffiWg)
wg.Done()
// respWaiters maps an in-flight call's resp pointer (a stable C-heap address)
// to the channel its result callback closes. We deliberately do NOT store a Go
// pointer (e.g. &sync.WaitGroup) in the C SdsResp: a goroutine parked in the
// call can have its stack moved by the GC, leaving the C-held Go pointer stale
// — a use-after-free that corrupts memory once the callback fires on the worker
// thread. Keying by the C pointer keeps all Go pointers on the Go side.
var respWaiters sync.Map // uintptr(resp) -> chan struct{}
// awaitResp allocates a resp, registers its waiter, runs `fire` (which
// dispatches the FFI call passing resp as userData), and blocks until the
// result callback closes the channel. The caller owns the returned resp and
// must C.freeResp it.
func awaitResp(fire func(resp unsafe.Pointer)) unsafe.Pointer {
resp := C.allocResp()
done := make(chan struct{})
respWaiters.Store(uintptr(resp), done)
fire(resp)
<-done
respWaiters.Delete(uintptr(resp))
return resp
}
// jsonByteArray marshals to a JSON array of byte values (e.g. [104,105]).
// libsds (nim-ffi) (de)serialises `seq[byte]` as a JSON number array, NOT
// base64 — Go's default []byte marshalling (base64 string) would not decode.
type jsonByteArray []byte
func (b jsonByteArray) MarshalJSON() ([]byte, error) {
if len(b) == 0 {
return []byte("[]"), nil
}
buf := make([]byte, 0, len(b)*4+2)
buf = append(buf, '[')
for i, v := range b {
if i > 0 {
buf = append(buf, ',')
}
buf = strconv.AppendUint(buf, uint64(v), 10)
}
buf = append(buf, ']')
return buf, nil
}
// Request payloads. Field names/types must match the {.ffi.} request objects
// declared in nim-sds' library/libsds.nim. Responses come back as JSON too;
// incoming []byte fields decode fine from a JSON number array.
type sdsConfig struct {
ParticipantId string `json:"participantId"`
}
type sdsWrapRequest struct {
Message jsonByteArray `json:"message"`
MessageId string `json:"messageId"`
ChannelId string `json:"channelId"`
}
type sdsWrapResponse struct {
Message []byte `json:"message"`
}
type sdsUnwrapRequest struct {
Message jsonByteArray `json:"message"`
}
type sdsMarkDependenciesRequest struct {
MessageIds []string `json:"messageIds"`
ChannelId string `json:"channelId"`
}
//export SdsGoCallback
func SdsGoCallback(ret C.int, msg *C.char, length C.size_t, resp unsafe.Pointer) {
if resp == nil {
return
}
m := (*C.SdsResp)(resp)
m.ret = ret
// libsds frees 'msg' as soon as this callback returns, so copy the bytes
// before unblocking the waiting goroutine that reads them.
if msg != nil && length > 0 {
m.msg = C.cGoMemDup(unsafe.Pointer(msg), length)
m.len = length
}
if ch, ok := respWaiters.Load(uintptr(resp)); ok {
close(ch.(chan struct{}))
}
}
func respString(resp unsafe.Pointer) string {
return C.GoStringN(C.getMyCharPtr(resp), C.int(C.getMyCharLen(resp)))
}
// request dispatches one FFI call and blocks until the result callback fires.
func (rm *ReliabilityManager) request(
errPrefix string,
fn func(resp unsafe.Pointer) C.int,
) (string, error) {
resp := awaitResp(func(r unsafe.Pointer) { fn(r) })
defer C.freeResp(resp)
if C.getRet(resp) != C.RET_OK {
return "", fmt.Errorf("%s: %s", errPrefix, respString(resp))
}
return respString(resp), nil
}
func NewReliabilityManager(logger *zap.Logger) (*ReliabilityManager, error) {
@ -174,22 +243,27 @@ func NewReliabilityManager(logger *zap.Logger) (*ReliabilityManager, error) {
rm.logger.Info("creating new reliability manager")
wg := sync.WaitGroup{}
// An empty participantId disables SDS-R, matching the previous behaviour.
configJson, err := json.Marshal(sdsConfig{ParticipantId: ""})
if err != nil {
return nil, fmt.Errorf("failed to encode config: %w", err)
}
cConfig := C.CString(string(configJson))
defer C.free(unsafe.Pointer(cConfig))
var resp = C.allocResp(unsafe.Pointer(&wg))
resp := awaitResp(func(r unsafe.Pointer) {
rm.rmCtx = C.cGoSdsCreate(cConfig, r)
})
defer C.freeResp(resp)
if C.getRet(resp) != C.RET_OK {
errMsg := C.GoStringN(C.getMyCharPtr(resp), C.int(C.getMyCharLen(resp)))
return nil, errors.New(errMsg)
if rm.rmCtx == nil || C.getRet(resp) != C.RET_OK {
return nil, fmt.Errorf("error creating reliability manager: %s", respString(resp))
}
wg.Add(1)
rm.rmCtx = C.cGoSdsNewReliabilityManager(resp)
wg.Wait()
C.cGoSdsSetEventCallback(rm.rmCtx)
// Register before wiring the event callback, since the callback routes by
// ctx through the registry.
registerReliabilityManager(rm)
C.cGoSdsSetEventCallback(rm.rmCtx)
C.cGoSdsSetRetrievalHintProvider(rm.rmCtx)
rm.logger.Debug("successfully created reliability manager")
@ -197,9 +271,9 @@ func NewReliabilityManager(logger *zap.Logger) (*ReliabilityManager, error) {
}
//export sdsGlobalEventCallback
func sdsGlobalEventCallback(callerRet C.int, msg *C.char, len C.size_t, userData unsafe.Pointer) {
msgStr := C.GoStringN(msg, C.int(len))
rm, ok := rmRegistry[userData] // userData contains rm's ctx
func sdsGlobalEventCallback(callerRet C.int, msg *C.char, length C.size_t, userData unsafe.Pointer) {
msgStr := C.GoStringN(msg, C.int(length))
rm, ok := lookupReliabilityManager(userData) // userData contains rm's ctx
if !ok {
return
}
@ -214,7 +288,7 @@ func sdsGlobalEventCallback(callerRet C.int, msg *C.char, len C.size_t, userData
//export sdsGlobalRetrievalHintProvider
func sdsGlobalRetrievalHintProvider(messageId *C.char, hint **C.char, hintLen *C.size_t, userData unsafe.Pointer) {
msgId := C.GoString(messageId)
rm, ok := rmRegistry[userData]
rm, ok := lookupReliabilityManager(userData)
if ok {
if rm.callbacks.RetrievalHintProvider != nil {
hintBytes := rm.callbacks.RetrievalHintProvider(MessageID(msgId))
@ -233,22 +307,16 @@ func (rm *ReliabilityManager) Cleanup() error {
rm.logger.Debug("cleaning up reliability manager")
wg := sync.WaitGroup{}
var resp = C.allocResp(unsafe.Pointer(&wg))
defer C.freeResp(resp)
wg.Add(1)
C.cGoSdsCleanupReliabilityManager(rm.rmCtx, resp)
wg.Wait()
if C.getRet(resp) == C.RET_OK {
unregisterReliabilityManager(rm)
rm.logger.Debug("cleaned up reliability manager")
return nil
_, err := rm.request("error CleanupReliabilityManager", func(resp unsafe.Pointer) C.int {
return C.cGoSdsDestroy(rm.rmCtx, resp)
})
if err != nil {
return err
}
errMsg := "error CleanupReliabilityManager: " + C.GoStringN(C.getMyCharPtr(resp), C.int(C.getMyCharLen(resp)))
return errors.New(errMsg)
unregisterReliabilityManager(rm)
rm.logger.Debug("cleaned up reliability manager")
return nil
}
func (rm *ReliabilityManager) Reset() error {
@ -258,21 +326,15 @@ func (rm *ReliabilityManager) Reset() error {
rm.logger.Debug("resetting reliability manager")
wg := sync.WaitGroup{}
var resp = C.allocResp(unsafe.Pointer(&wg))
defer C.freeResp(resp)
wg.Add(1)
C.cGoSdsResetReliabilityManager(rm.rmCtx, resp)
wg.Wait()
if C.getRet(resp) == C.RET_OK {
rm.logger.Debug("successfully resetted reliability manager")
return nil
_, err := rm.request("error ResetReliabilityManager", func(resp unsafe.Pointer) C.int {
return C.cGoSdsReset(rm.rmCtx, resp)
})
if err != nil {
return err
}
errMsg := "error ResetReliabilityManager: " + C.GoStringN(C.getMyCharPtr(resp), C.int(C.getMyCharLen(resp)))
return errors.New(errMsg)
rm.logger.Debug("successfully resetted reliability manager")
return nil
}
func (rm *ReliabilityManager) WrapOutgoingMessage(message []byte, messageId MessageID, channelId string) ([]byte, error) {
@ -281,55 +343,33 @@ func (rm *ReliabilityManager) WrapOutgoingMessage(message []byte, messageId Mess
}
logger := rm.logger.With(zap.String("messageId", string(messageId)))
logger.Debug("wrapping outgoing message", zap.String("messageId", string(messageId)))
logger.Debug("wrapping outgoing message")
wg := sync.WaitGroup{}
var resp = C.allocResp(unsafe.Pointer(&wg))
defer C.freeResp(resp)
cMessageId := C.CString(string(messageId))
defer C.free(unsafe.Pointer(cMessageId))
var cMessagePtr unsafe.Pointer
if len(message) > 0 {
cMessagePtr = C.CBytes(message) // C.CBytes allocates memory that needs to be freed
defer C.free(cMessagePtr)
} else {
cMessagePtr = nil
reqJson, err := json.Marshal(sdsWrapRequest{
Message: message,
MessageId: string(messageId),
ChannelId: channelId,
})
if err != nil {
return nil, fmt.Errorf("failed to encode wrap request: %w", err)
}
cMessageLen := C.size_t(len(message))
cReq := C.CString(string(reqJson))
defer C.free(unsafe.Pointer(cReq))
cChannelId := C.CString(channelId)
defer C.free(unsafe.Pointer(cChannelId))
wg.Add(1)
C.cGoSdsWrapOutgoingMessage(rm.rmCtx, cMessagePtr, cMessageLen, cMessageId, cChannelId, resp)
wg.Wait()
if C.getRet(resp) == C.RET_OK {
resStr := C.GoStringN(C.getMyCharPtr(resp), C.int(C.getMyCharLen(resp)))
if resStr == "" {
logger.Debug("received empty res string for messageId")
return nil, nil
}
logger.Debug("successfully wrapped message")
parts := strings.Split(resStr, ",")
bytes := make([]byte, len(parts))
for i, part := range parts {
n, err := strconv.Atoi(strings.TrimSpace(part))
if err != nil {
panic(err)
}
bytes[i] = byte(n)
}
return bytes, nil
respJson, err := rm.request("error WrapOutgoingMessage", func(resp unsafe.Pointer) C.int {
return C.cGoSdsWrapOutgoingMessage(rm.rmCtx, cReq, resp)
})
if err != nil {
return nil, err
}
errMsg := "error WrapOutgoingMessage: " + C.GoStringN(C.getMyCharPtr(resp), C.int(C.getMyCharLen(resp)))
return nil, errors.New(errMsg)
var wrapResp sdsWrapResponse
if err := json.Unmarshal([]byte(respJson), &wrapResp); err != nil {
return nil, fmt.Errorf("failed to decode wrap response: %w", err)
}
logger.Debug("successfully wrapped message")
return wrapResp.Message, nil
}
func (rm *ReliabilityManager) UnwrapReceivedMessage(message []byte) (*UnwrappedMessage, error) {
@ -337,43 +377,27 @@ func (rm *ReliabilityManager) UnwrapReceivedMessage(message []byte) (*UnwrappedM
return nil, errEmptyReliabilityManager
}
wg := sync.WaitGroup{}
var resp = C.allocResp(unsafe.Pointer(&wg))
defer C.freeResp(resp)
var cMessagePtr unsafe.Pointer
if len(message) > 0 {
cMessagePtr = C.CBytes(message) // C.CBytes allocates memory that needs to be freed
defer C.free(cMessagePtr)
} else {
cMessagePtr = nil
reqJson, err := json.Marshal(sdsUnwrapRequest{Message: message})
if err != nil {
return nil, fmt.Errorf("failed to encode unwrap request: %w", err)
}
cMessageLen := C.size_t(len(message))
cReq := C.CString(string(reqJson))
defer C.free(unsafe.Pointer(cReq))
wg.Add(1)
C.cGoSdsUnwrapReceivedMessage(rm.rmCtx, cMessagePtr, cMessageLen, resp)
wg.Wait()
if C.getRet(resp) == C.RET_OK {
resStr := C.GoStringN(C.getMyCharPtr(resp), C.int(C.getMyCharLen(resp)))
if resStr == "" {
rm.logger.Debug("received empty res string")
return nil, nil
}
rm.logger.Debug("successfully unwrapped message")
rm.logger.Debug("Unwrapped message JSON: %s", resStr)
var unwrappedMessage UnwrappedMessage
err := json.Unmarshal([]byte(resStr), &unwrappedMessage)
if err != nil {
return nil, errorspkg.Wrap(err, "failed to unmarshal unwrapped message")
}
return &unwrappedMessage, nil
respJson, err := rm.request("error UnwrapReceivedMessage", func(resp unsafe.Pointer) C.int {
return C.cGoSdsUnwrapReceivedMessage(rm.rmCtx, cReq, resp)
})
if err != nil {
return nil, err
}
errMsg := "error UnwrapReceivedMessage: " + C.GoStringN(C.getMyCharPtr(resp), C.int(C.getMyCharLen(resp)))
return nil, errors.New(errMsg)
var unwrappedMessage UnwrappedMessage
if err := json.Unmarshal([]byte(respJson), &unwrappedMessage); err != nil {
return nil, fmt.Errorf("failed to decode unwrap response: %w", err)
}
rm.logger.Debug("successfully unwrapped message")
return &unwrappedMessage, nil
}
// MarkDependenciesMet informs the library that dependencies are met
@ -386,40 +410,26 @@ func (rm *ReliabilityManager) MarkDependenciesMet(messageIDs []MessageID, channe
return nil // Nothing to do
}
wg := sync.WaitGroup{}
var resp = C.allocResp(unsafe.Pointer(&wg))
defer C.freeResp(resp)
// Convert Go string slice to C array of C strings (char**)
cMessageIDs := make([]*C.char, len(messageIDs))
ids := make([]string, len(messageIDs))
for i, id := range messageIDs {
cMessageIDs[i] = C.CString(string(id))
defer C.free(unsafe.Pointer(cMessageIDs[i])) // Ensure each CString is freed
ids[i] = string(id)
}
reqJson, err := json.Marshal(sdsMarkDependenciesRequest{MessageIds: ids, ChannelId: channelId})
if err != nil {
return fmt.Errorf("failed to encode mark-dependencies request: %w", err)
}
cReq := C.CString(string(reqJson))
defer C.free(unsafe.Pointer(cReq))
_, err = rm.request("error MarkDependenciesMet", func(resp unsafe.Pointer) C.int {
return C.cGoSdsMarkDependenciesMet(rm.rmCtx, cReq, resp)
})
if err != nil {
return err
}
// Create a pointer (**C.char) to the first element of the slice
var cMessageIDsPtr **C.char
if len(cMessageIDs) > 0 {
cMessageIDsPtr = &cMessageIDs[0]
} else {
cMessageIDsPtr = nil // Handle empty slice case
}
wg.Add(1)
cChannelId := C.CString(channelId)
defer C.free(unsafe.Pointer(cChannelId))
// Pass the pointer variable (cMessageIDsPtr) directly, which is of type **C.char
C.cGoSdsMarkDependenciesMet(rm.rmCtx, cMessageIDsPtr, C.size_t(len(messageIDs)), cChannelId, resp)
wg.Wait()
if C.getRet(resp) == C.RET_OK {
rm.logger.Debug("successfully marked dependencies as met")
return nil
}
errMsg := "error MarkDependenciesMet: " + C.GoStringN(C.getMyCharPtr(resp), C.int(C.getMyCharLen(resp)))
return errors.New(errMsg)
rm.logger.Debug("successfully marked dependencies as met")
return nil
}
func (rm *ReliabilityManager) StartPeriodicTasks() error {
@ -429,19 +439,13 @@ func (rm *ReliabilityManager) StartPeriodicTasks() error {
rm.logger.Debug("starting periodic tasks")
wg := sync.WaitGroup{}
var resp = C.allocResp(unsafe.Pointer(&wg))
defer C.freeResp(resp)
wg.Add(1)
C.cGoSdsStartPeriodicTasks(rm.rmCtx, resp)
wg.Wait()
if C.getRet(resp) == C.RET_OK {
rm.logger.Debug("successfully started periodic tasks")
return nil
_, err := rm.request("error StartPeriodicTasks", func(resp unsafe.Pointer) C.int {
return C.cGoSdsStartPeriodicTasks(rm.rmCtx, resp)
})
if err != nil {
return err
}
errMsg := "error StartPeriodicTasks: " + C.GoStringN(C.getMyCharPtr(resp), C.int(C.getMyCharLen(resp)))
return errors.New(errMsg)
rm.logger.Debug("successfully started periodic tasks")
return nil
}

44
sds/sds_common.go
View File

@ -2,6 +2,7 @@ package sds
import (
"encoding/json"
"sync"
"time"
"unsafe"
@ -16,6 +17,7 @@ type EventCallbacks struct {
OnMessageSent func(messageId MessageID, channelId string)
OnMissingDependencies func(messageId MessageID, missingDeps []HistoryEntry, channelId string)
OnPeriodicSync func()
OnRepairReady func(message []byte, channelId string)
RetrievalHintProvider func(messageId MessageID) []byte
}
@ -33,11 +35,27 @@ type ReliabilityManager struct {
// be invoked depending on the ctx received
var rmRegistry map[unsafe.Pointer]*ReliabilityManager
// rmRegistryMu guards rmRegistry. libsds invokes the global callbacks on its
// own worker/event threads, so reads from those callbacks race the register/
// unregister writes done on the goroutines that create and clean up managers.
var rmRegistryMu sync.RWMutex
func init() {
rmRegistry = make(map[unsafe.Pointer]*ReliabilityManager)
}
// lookupReliabilityManager resolves the manager for a ctx under the read lock.
// Used by the global callbacks, which run on libsds threads.
func lookupReliabilityManager(ctx unsafe.Pointer) (*ReliabilityManager, bool) {
rmRegistryMu.RLock()
defer rmRegistryMu.RUnlock()
rm, ok := rmRegistry[ctx]
return rm, ok
}
func registerReliabilityManager(rm *ReliabilityManager) {
rmRegistryMu.Lock()
defer rmRegistryMu.Unlock()
_, ok := rmRegistry[rm.rmCtx]
if !ok {
rmRegistry[rm.rmCtx] = rm
@ -45,6 +63,8 @@ func registerReliabilityManager(rm *ReliabilityManager) {
}
func unregisterReliabilityManager(rm *ReliabilityManager) {
rmRegistryMu.Lock()
defer rmRegistryMu.Unlock()
delete(rmRegistry, rm.rmCtx)
}
@ -58,11 +78,18 @@ type msgEvent struct {
}
type missingDepsEvent struct {
MessageId MessageID `json:"messageId"`
MessageId MessageID `json:"messageId"`
MissingDeps []HistoryEntry `json:"missingDeps"`
ChannelId string `json:"channelId"`
}
// repairReadyEvent mirrors libsds' repair_ready JSON payload. nim-sds base64-
// encodes the message bytes, which encoding/json decodes back into []byte.
type repairReadyEvent struct {
Message []byte `json:"message"`
ChannelId string `json:"channelId"`
}
func (rm *ReliabilityManager) RegisterCallbacks(callbacks EventCallbacks) {
rm.callbacks = callbacks
}
@ -86,6 +113,8 @@ func (rm *ReliabilityManager) OnEvent(eventStr string) {
if rm.callbacks.OnPeriodicSync != nil {
rm.callbacks.OnPeriodicSync()
}
case "repair_ready":
rm.parseRepairReadyEvent(eventStr)
}
}
@ -132,3 +161,16 @@ func (rm *ReliabilityManager) parseMissingDepsEvent(eventStr string) {
rm.callbacks.OnMissingDependencies(missingDepsEvent.MessageId, missingDepsEvent.MissingDeps, missingDepsEvent.ChannelId)
}
}
func (rm *ReliabilityManager) parseRepairReadyEvent(eventStr string) {
repairReadyEvent := repairReadyEvent{}
err := json.Unmarshal([]byte(eventStr), &repairReadyEvent)
if err != nil {
rm.logger.Error("failed to parse repair ready event", zap.Error(err))
return
}
if rm.callbacks.OnRepairReady != nil {
rm.callbacks.OnRepairReady(repairReadyEvent.Message, repairReadyEvent.ChannelId)
}
}