queued disk stuff

pkg/actor/disk_storage.go

@@ -6,13 +6,22 @@ import (
 	"log"
 	"os"
 	"path/filepath"
+	"sync"
+	"time"
 
 	"github.com/gogo/protobuf/proto"
 )
 
+type QueueEvent struct {
+	TimeStamp time.Time
+	Message   proto.Message
+}
+
 type DiskStorage[V any] struct {
 	*StateStorage
-	path string
+	path  string
+	done  chan struct{}
+	queue *sync.Map // map[uint64][]QueueEvent
 }
 
 type LogStorage[V any] interface {
@@ -20,13 +29,51 @@ type LogStorage[V any] interface {
 	AppendEvent(id uint64, msg proto.Message) error
 }
 
-func NewDiskStorage[V any](path string, registry MutationRegistry) LogStorage[V] {
+func NewDiskStorage[V any](path string, registry MutationRegistry) *DiskStorage[V] {
 	return &DiskStorage[V]{
 		StateStorage: NewState(registry),
 		path:         path,
+		done:         make(chan struct{}),
 	}
 }
 
+func (s *DiskStorage[V]) SaveLoop(duration time.Duration) {
+	s.queue = &sync.Map{}
+	ticker := time.NewTicker(duration)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-s.done:
+			s.save()
+			return
+		case <-ticker.C:
+			s.save()
+		}
+	}
+}
+
+func (s *DiskStorage[V]) save() {
+	s.queue.Range(func(key, value any) bool {
+		id := key.(uint64)
+		path := s.logPath(id)
+		fh, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+		if err != nil {
+			log.Printf("failed to open event log file: %v", err)
+			return true
+		}
+		defer fh.Close()
+
+		if qe, ok := value.([]QueueEvent); ok {
+			for _, msg := range qe {
+				if err := s.Append(fh, msg.Message, msg.TimeStamp); err != nil {
+					log.Printf("failed to append event to log file: %v", err)
+				}
+			}
+		}
+		return true
+	})
+}
+
 func (s *DiskStorage[V]) logPath(id uint64) string {
 	return filepath.Join(s.path, fmt.Sprintf("%d.events.log", id))
 }
@@ -48,15 +95,33 @@ func (s *DiskStorage[V]) LoadEvents(id uint64, grain Grain[V]) error {
 	})
 }
 
-func (s *DiskStorage[V]) AppendEvent(id uint64, msg proto.Message) error {
-	path := s.logPath(id)
-	fh, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
-	if err != nil {
-		log.Printf("failed to open event log file: %v", err)
-		return err
-	}
-	defer fh.Close()
+func (s *DiskStorage[V]) Close() {
+	s.save()
+	close(s.done)
+}
 
-	return s.Append(fh, msg)
+func (s *DiskStorage[V]) AppendEvent(id uint64, msg proto.Message) error {
+	if s.queue != nil {
+
+		queue := make([]QueueEvent, 0)
+		data, found := s.queue.Load(id)
+		if found {
+			queue = data.([]QueueEvent)
+		}
+		queue = append(queue, QueueEvent{Message: msg})
+		s.queue.Store(id, queue)
+		return nil
+	} else {
+
+		path := s.logPath(id)
+		fh, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+		if err != nil {
+			log.Printf("failed to open event log file: %v", err)
+			return err
+		}
+		defer fh.Close()
+
+		return s.Append(fh, msg, time.Now())
+	}
 
 }

pkg/actor/mutation_registry_test.go

@@ -3,6 +3,7 @@ package actor
 import (
 	"errors"
 	"reflect"
+	"slices"
 	"testing"
 
 	"git.tornberg.me/go-cart-actor/pkg/messages"
@@ -16,7 +17,7 @@ type cartState struct {
 func TestRegisteredMutationBasics(t *testing.T) {
 	reg := NewMutationRegistry().(*ProtoMutationRegistry)
 
-	addItemMutation := NewMutation[cartState, *messages.AddItem](
+	addItemMutation := NewMutation(
 		func(state *cartState, msg *messages.AddItem) error {
 			state.calls++
 			// copy to avoid external mutation side-effects (not strictly necessary for the test)
@@ -39,13 +40,13 @@ func TestRegisteredMutationBasics(t *testing.T) {
 
 	// RegisteredMutations: membership (order not guaranteed)
 	names := reg.RegisteredMutations()
-	if !stringSliceContains(names, "AddItem") {
+	if !slices.Contains(names, "AddItem") {
 		t.Fatalf("RegisteredMutations missing AddItem, got %v", names)
 	}
 
 	// RegisteredMutationTypes: membership (order not guaranteed)
 	types := reg.RegisteredMutationTypes()
-	if !typeSliceContains(types, reflect.TypeOf(messages.AddItem{})) {
+	if !slices.Contains(types, reflect.TypeOf(messages.AddItem{})) {
 		t.Fatalf("RegisteredMutationTypes missing AddItem type, got %v", types)
 	}
 
@@ -131,21 +132,3 @@ func TestRegisteredMutationBasics(t *testing.T) {
 // }
 
 // Helpers
-
-func stringSliceContains(list []string, target string) bool {
-	for _, s := range list {
-		if s == target {
-			return true
-		}
-	}
-	return false
-}
-
-func typeSliceContains(list []reflect.Type, target reflect.Type) bool {
-	for _, t := range list {
-		if t == target {
-			return true
-		}
-	}
-	return false
-}

pkg/actor/state.go

@@ -48,14 +48,14 @@ func (s *StateStorage) Load(r io.Reader, onMessage func(msg proto.Message)) erro
 	return err
 }
 
-func (s *StateStorage) Append(io io.Writer, mutation proto.Message) error {
+func (s *StateStorage) Append(io io.Writer, mutation proto.Message, timeStamp time.Time) error {
 	typeName, ok := s.registry.GetTypeName(mutation)
 	if !ok {
 		return ErrUnknownType
 	}
 	event := &StorageEvent{
 		Type:      typeName,
-		TimeStamp: time.Now(),
+		TimeStamp: timeStamp,
 		Mutation:  mutation,
 	}
 	jsonBytes, err := json.Marshal(event)

pkg/discovery/discovery.go

@@ -2,6 +2,7 @@ package discovery
 
 import (
 	"context"
+	"log"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -46,7 +47,7 @@ func (k *K8sDiscovery) Watch() (<-chan HostChange, error) {
 			TimeoutSeconds: &timeout,
 		})
 	}
-	watcher, err := toolsWatch.NewRetryWatcher("1", &cache.ListWatch{WatchFunc: watcherFn})
+	watcher, err := toolsWatch.NewRetryWatcherWithContext(k.ctx, "1", &cache.ListWatch{WatchFunc: watcherFn})
 	if err != nil {
 		return nil, err
 	}
@@ -55,6 +56,7 @@ func (k *K8sDiscovery) Watch() (<-chan HostChange, error) {
 		for event := range watcher.ResultChan() {
 
 			pod := event.Object.(*v1.Pod)
+			log.Printf("pod change %v", pod.Status)
 			ch <- HostChange{
 				Host: pod.Status.PodIP,
 				Type: event.Type,