Verified feat: add on pebble state backend and using sync Pool to reuse buffer []byte

7 files changed, 137 insertions, 20 deletions

diff --git a/README.md b/README.md
new file mode 100644
index 0000000..a180257
--- /dev/null
+++ b/README.md
@@ -0,0 +1,7 @@
+Flux 
+====
+
+Flux is an event-driven stream processing engine written in Go, 
+designed from first principles to handle stateful stream computation,
+out-of-order event time processing, high-throughput windowing, 
+and distributed fault-tolerant state backend.
diff --git a/core/pool.go b/core/pool.go
new file mode 100644
index 0000000..484b342
--- /dev/null
+++ b/core/pool.go
@@ -0,0 +1,38 @@
+package core
+
+import "sync"
+
+
+const (
+        DefaultBufferSize = 1024
+        MaxBufferSize = 64 * 1024
+)
+
+var bytesPool = sync.Pool{
+        New: func () any {
+                b := make([]byte, DefaultBufferSize)
+                return &b
+        },
+}
+
+func GetBuffer(size int) []byte {
+        bufPtr := bytesPool.Get().(*[]byte)
+
+        buf := *bufPtr
+
+
+        if size > cap(buf) {
+                buf = make([]byte, size)
+        }
+
+        return buf[:size]
+}
+
+
+func PutBuffer(buf []byte) {
+        if cap(buf) < MaxBufferSize{
+                // reset
+                buf = buf[:0]
+                bytesPool.Put(&buf)
+        }
+}
diff --git a/core/state_backend.go b/core/state_backend.go
index 5870739..ae9fe1d 100644
--- a/core/state_backend.go
+++ b/core/state_backend.go
@@ -2,6 +2,9 @@ package core
 
 import (
         "context"
+        "encoding/json"
+        "fmt"
+
 )
 
 type StateBackend interface {
@@ -22,6 +25,7 @@ type stateBackendKeyType struct{}
 
 var stateBackendKey stateBackendKeyType
 
+// type adapter
 type typedValueState[T any] struct {
         inner ValueState[any]
 }
@@ -32,14 +36,24 @@ func CastValueType[T any](inner ValueState[any]) ValueState[T] {
 
 func (s *typedValueState[T]) Value(ctx context.Context) (T, error) {
         val, err := s.inner.Value(ctx)
+        var zero T
         if err != nil {
-                var zero T
                 return zero, err
         }
         if val == nil {
-                var zero T
                 return zero, nil
         }
+
+        if rawBytes, ok := val.([]byte); ok {
+                defer PutBuffer(rawBytes)
+
+                var result T
+                if err := json.Unmarshal(rawBytes, &result); err != nil {
+                        return zero, fmt.Errorf("failed to unmarshal state: %w", err)
+                }
+                return result, nil
+        }
+
         return val.(T), nil
 }
 
diff --git a/go.mod b/go.mod
index df5991f..a99ec15 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
 module flux
 
-go 1.26.3
+go 1.27
 
 require github.com/cockroachdb/pebble/v2 v2.1.7
 
diff --git a/state/memory_backend.go b/state/memory_backend.go
index dda577e..cce8194 100644
--- a/state/memory_backend.go
+++ b/state/memory_backend.go
@@ -39,15 +39,15 @@ func (ms *MemoryStateBackend) SetCurrentKey(key string) error {
         return nil
 }
 
-func (ms *MemoryStateBackend) GetValueState(ctx context.Context, statename string) (core.ValueState[any], error) {
+func (ms *MemoryStateBackend) GetValueState(ctx context.Context, stateName string) (core.ValueState[any], error) {
         // Assign if state not existed
         
-        if _, ok := ms.states[statename]; !ok {
+        if _, ok := ms.states[stateName]; !ok {
                 // create state name map if not existed 
-                ms.states[statename] = make(map[int]map[string]map[any]any)
+                ms.states[stateName] = make(map[int]map[string]map[any]any)
         }
 
-        return &memoryStateValue{backend: ms, stateName: statename}, nil
+        return &memoryStateValue{backend: ms, stateName: stateName}, nil
 }
 
 func (ms *MemoryStateBackend) GetCurrentNamespace() any {
diff --git a/state/pebble_backend.go b/state/pebble_backend.go
index e428988..fbb8d68 100644
--- a/state/pebble_backend.go
+++ b/state/pebble_backend.go
@@ -33,10 +33,11 @@ func NewPebbleStateBackend(dir string, maxParalleism int) (*PebbleStateBackend,
                 return nil, fmt.Errorf("failed to open pebbledb %w", err)
         }
         return &PebbleStateBackend{
-                maxParallelism: maxParalleism,
-                db:             db,
-                stateMap:       make(map[string]uint8),
-                nextStateID:    1,
+                maxParallelism:   maxParalleism,
+                db:               db,
+                stateMap:         make(map[string]uint8),
+                nextStateID:      1,
+                currentNamespace: &core.VoidNamespace{},
         }, nil
 }
 
@@ -60,23 +61,27 @@ func (p *PebbleStateBackend) GetCurrentNamespace() any {
 }
 
 // GetValueState(ctx context.Context, key string) (ValueState[any], error)
-func (p *PebbleStateBackend) GetValueState(ctx context.Context, statename string) (core.ValueState[any], error) {
-        stateID := p.getStateID(statename)
-        return nil, nil
+func (p *PebbleStateBackend) GetValueState(ctx context.Context, stateName string) (core.ValueState[any], error) {
+        stateID := p.getStateID(stateName)
+
+        return &pebbleStateValue{
+                backend: p,
+                stateID: stateID,
+        }, nil
 }
 
-func (p *PebbleStateBackend) getStateID(statename string) uint8 {
+func (p *PebbleStateBackend) getStateID(stateName string) uint8 {
         p.mu.Lock()
         defer p.mu.Unlock()
 
-        if _, ok := p.stateMap[statename]; !ok {
+        if _, ok := p.stateMap[stateName]; !ok {
 
                 id := p.nextStateID
-                p.stateMap[statename] = id
+                p.stateMap[stateName] = id
                 p.nextStateID++
         }
 
-        return p.stateMap[statename]
+        return p.stateMap[stateName]
 }
 
 func (p *PebbleStateBackend) calculateNamespaceSize() int {
@@ -94,7 +99,7 @@ func (p *PebbleStateBackend) calculateNamespaceSize() int {
         return nsLen
 }
 
-func (p *PebbleStateBackend) deserialize(statename string) []byte {
+func (p *PebbleStateBackend) deserialize(stateID uint8) []byte {
         // this will store in SSTable, order is important, keygroup MUST be first for scaling
         // 2B(KeyGroup) + 1B(StateID) + 2B(KeyLen) + len(Key) + 16B(Window)
 
@@ -105,7 +110,6 @@ func (p *PebbleStateBackend) deserialize(statename string) []byte {
         b = binary.BigEndian.AppendUint16(b, uint16(p.currentGroupKey))
 
         // add state id
-        stateID := p.getStateID(statename)
         b = append(b, stateID)
 
         // add key
diff --git a/state/pebble_state.go b/state/pebble_state.go
new file mode 100644
index 0000000..3eeaa05
--- /dev/null
+++ b/state/pebble_state.go
@@ -0,0 +1,54 @@
+package state
+
+import (
+        "context"
+        "encoding/json"
+        "errors"
+        "flux/core"
+
+        "github.com/cockroachdb/pebble/v2"
+)
+
+type pebbleStateValue struct {
+        backend *PebbleStateBackend
+        stateID uint8
+}
+
+func (p *pebbleStateValue) Update(ctx context.Context, newVal any) (any, error) {
+        key := p.backend.deserialize(p.stateID)
+        // convert newVal to []byte raw
+        v, err := json.Marshal(newVal)
+        if err != nil {
+                return nil, err
+        }
+
+        err = p.backend.db.Set(key, v, pebble.NoSync)
+        if err != nil {
+                return nil, err
+        }
+
+        return newVal, nil
+}
+
+func (p *pebbleStateValue) Value(ctx context.Context) (any, error) {
+        key := p.backend.deserialize(p.stateID)
+        v, closer, err := p.backend.db.Get(key)
+        if err != nil {
+                if errors.Is(err, pebble.ErrNotFound) {
+                        return nil, nil 
+                }
+                return nil, err
+        }
+
+        defer closer.Close()
+
+        res := core.GetBuffer(len(v))
+        copy(res, v)
+        return v, nil
+}
+
+func (p *pebbleStateValue) Clear(ctx context.Context) error {
+        key := p.backend.deserialize(p.stateID)
+
+        return p.backend.db.Delete(key, pebble.NoSync)
+}