revert to legacy autoack events

Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de>

services/activitylog/pkg/command/server.go

@@ -28,7 +28,7 @@ import (
 	svcEvents "github.com/opencloud-eu/opencloud/services/activitylog/pkg/service/events"
 	svcHttp "github.com/opencloud-eu/opencloud/services/activitylog/pkg/service/http"
 	"github.com/opencloud-eu/reva/v2/pkg/events"
-	"github.com/opencloud-eu/reva/v2/pkg/events/raw"
+	"github.com/opencloud-eu/reva/v2/pkg/events/stream"
 	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
 )
 
@@ -156,7 +156,7 @@ func Server(cfg *config.Config) *cobra.Command {
 			if !cfg.Events.Disabled {
 
 				connName := generators.GenerateConnectionName(cfg.Service.Name, generators.NTypeBus)
-				evStream, err := raw.FromConfig(ctx, connName, raw.Config{
+				evStream, err := stream.NatsFromConfig(connName, false, stream.NatsConfig{
 					Endpoint:             cfg.Events.Endpoint,
 					Cluster:              cfg.Events.Cluster,
 					EnableTLS:            cfg.Events.EnableTLS,
@@ -164,8 +164,6 @@ func Server(cfg *config.Config) *cobra.Command {
 					TLSRootCACertificate: cfg.Events.TLSRootCACertificate,
 					AuthUsername:         cfg.Events.AuthUsername,
 					AuthPassword:         cfg.Events.AuthPassword,
-					MaxAckPending:        cfg.Events.MaxAckPending,
-					AckWait:              cfg.Events.AckWait,
 				})
 				if err != nil {
 					logger.Error().Err(err).Msg("Failed to initialize event stream")
@@ -187,11 +185,11 @@ func Server(cfg *config.Config) *cobra.Command {
 					return err
 				}
 
-				gr.Add(runner.New(cfg.Service.Name+".svc", func() error {
-					return eventSvc.Run()
-				}, func() {
-					eventSvc.Close()
-				}))
+			gr.Add(runner.New(cfg.Service.Name+".svc", func() error {
+				return eventSvc.Run()
+			}, func() {
+				eventSvc.Close()
+			}))
 			} else {
 				logger.Info().Msg("event listening disabled, not starting event service")
 			}

services/activitylog/pkg/config/config.go

@@ -40,16 +40,14 @@ type Config struct {
 
 // Events combines the configuration options for the event bus.
 type Events struct {
-	Disabled             bool          `yaml:"disabled" env:"ACTIVITYLOG_EVENTS_DISABLED" desc:"Disables listening for events. Set this to true if the service should only handle HTTP requests." introductionVersion:"%NEXT%"`
-	Endpoint             string        `yaml:"endpoint" env:"OC_EVENTS_ENDPOINT" desc:"The address of the event system. The event system is the message queuing service. It is used as message broker for the microservice architecture." introductionVersion:"1.0.0"`
-	Cluster              string        `yaml:"cluster" env:"OC_EVENTS_CLUSTER" desc:"The clusterID of the event system. The event system is the message queuing service. It is used as message broker for the microservice architecture. Mandatory when using NATS as event system." introductionVersion:"1.0.0"`
-	TLSInsecure          bool          `yaml:"tls_insecure" env:"OC_INSECURE;OC_EVENTS_TLS_INSECURE" desc:"Whether to verify the server TLS certificates." introductionVersion:"1.0.0"`
-	TLSRootCACertificate string        `yaml:"tls_root_ca_certificate" env:"OC_EVENTS_TLS_ROOT_CA_CERTIFICATE" desc:"The root CA certificate used to validate the server's TLS certificate. If provided NOTIFICATIONS_EVENTS_TLS_INSECURE will be seen as false." introductionVersion:"1.0.0"`
-	EnableTLS            bool          `yaml:"enable_tls" env:"OC_EVENTS_ENABLE_TLS" desc:"Enable TLS for the connection to the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"1.0.0"`
-	AuthUsername         string        `yaml:"username" env:"OC_EVENTS_AUTH_USERNAME" desc:"The username to authenticate with the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"1.0.0"`
-	AuthPassword         string        `yaml:"password" env:"OC_EVENTS_AUTH_PASSWORD" desc:"The password to authenticate with the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"1.0.0"`
-	MaxAckPending        int           `yaml:"max_ack_pending" env:"ACTIVITYLOG_EVENTS_MAX_ACK_PENDING" desc:"The maximum number of unacknowledged messages. This is used to limit the number of messages that can be in flight at the same time." introductionVersion:"%NEXT%"`
-	AckWait              time.Duration `yaml:"ack_wait" env:"ACTIVITYLOG_EVENTS_ACK_WAIT" desc:"The time to wait for an ack before the message is redelivered. This is used to ensure that messages are not lost if the consumer crashes." introductionVersion:"%NEXT%"`
+	Disabled             bool   `yaml:"disabled" env:"ACTIVITYLOG_EVENTS_DISABLED" desc:"Disables listening for events. Set this to true if the service should only handle HTTP requests." introductionVersion:"%NEXT%"`
+	Endpoint             string `yaml:"endpoint" env:"OC_EVENTS_ENDPOINT" desc:"The address of the event system. The event system is the message queuing service. It is used as message broker for the microservice architecture." introductionVersion:"1.0.0"`
+	Cluster              string `yaml:"cluster" env:"OC_EVENTS_CLUSTER" desc:"The clusterID of the event system. The event system is the message queuing service. It is used as message broker for the microservice architecture. Mandatory when using NATS as event system." introductionVersion:"1.0.0"`
+	TLSInsecure          bool   `yaml:"tls_insecure" env:"OC_INSECURE;OC_EVENTS_TLS_INSECURE" desc:"Whether to verify the server TLS certificates." introductionVersion:"1.0.0"`
+	TLSRootCACertificate string `yaml:"tls_root_ca_certificate" env:"OC_EVENTS_TLS_ROOT_CA_CERTIFICATE" desc:"The root CA certificate used to validate the server's TLS certificate. If provided NOTIFICATIONS_EVENTS_TLS_INSECURE will be seen as false." introductionVersion:"1.0.0"`
+	EnableTLS            bool   `yaml:"enable_tls" env:"OC_EVENTS_ENABLE_TLS" desc:"Enable TLS for the connection to the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"1.0.0"`
+	AuthUsername         string `yaml:"username" env:"OC_EVENTS_AUTH_USERNAME" desc:"The username to authenticate with the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"1.0.0"`
+	AuthPassword         string `yaml:"password" env:"OC_EVENTS_AUTH_PASSWORD" desc:"The password to authenticate with the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"1.0.0"`
 }
 
 // Store configures the store to use

services/activitylog/pkg/config/defaults/defaultconfig.go

@@ -29,11 +29,9 @@ func DefaultConfig() *config.Config {
 			Name: "activitylog",
 		},
 		Events: config.Events{
-			Endpoint:      "127.0.0.1:9233",
-			Cluster:       "opencloud-cluster",
-			EnableTLS:     false,
-			MaxAckPending: 1000,
-			AckWait:       1 * time.Minute,
+			Endpoint:  "127.0.0.1:9233",
+			Cluster:   "opencloud-cluster",
+			EnableTLS: false,
 		},
 		Store: config.Store{
 			Store:    "nats-js-kv",

services/activitylog/pkg/service/activitylog/activitylog.go

@@ -77,7 +77,7 @@ func New(kv nats.KeyValue, opts ...Option) (*ActivityLog, error) {
 		maxActivities: o.MaxActivities,
 		natskv:        kv,
 	}
-	s.debouncer = NewDebouncer(o.Logger, o.WriteBufferDuration, s.StoreActivity)
+	s.debouncer = NewDebouncer(o.WriteBufferDuration, s.StoreActivity)
 
 	// run migrations
 	err = s.runMigrations(context.Background(), kv)
@@ -89,7 +89,7 @@ func New(kv nats.KeyValue, opts ...Option) (*ActivityLog, error) {
 }
 
 // RemoveResource removes the resource from the store
-func (a *ActivityLog) RemoveResource(rid *provider.ResourceId, ack func() error) error {
+func (a *ActivityLog) RemoveResource(rid *provider.ResourceId) error {
 	if rid == nil {
 		return fmt.Errorf("resource id is required")
 	}
@@ -101,16 +101,10 @@ func (a *ActivityLog) RemoveResource(rid *provider.ResourceId, ack func() error)
 	if err != nil {
 		return fmt.Errorf("could not delete resource %s: %w", rid.OpaqueId, err)
 	}
-	go func() {
-		err := ack()
-		if err != nil {
-			a.log.Error().Err(err).Msg("error while acknowledging resource removal")
-		}
-	}()
 	return nil
 }
 
-func (a *ActivityLog) AddActivity(ctx context.Context, initRef *provider.Reference, parentId *provider.ResourceId, eventID string, timestamp time.Time, getResource func(context.Context, *provider.Reference) (*provider.ResourceInfo, error), ack func() error) error {
+func (a *ActivityLog) AddActivity(ctx context.Context, initRef *provider.Reference, parentId *provider.ResourceId, eventID string, timestamp time.Time, getResource func(context.Context, *provider.Reference) (*provider.ResourceInfo, error)) error {
 	var (
 		err   error
 		depth int
@@ -141,7 +135,7 @@ func (a *ActivityLog) AddActivity(ctx context.Context, initRef *provider.Referen
 			EventID:   eventID,
 			Depth:     depth,
 			Timestamp: timestamp,
-		}, ack)
+		})
 
 		if id.OpaqueId == id.SpaceId {
 			// we are at the root of the space, no need to go further

services/activitylog/pkg/service/activitylog/activitylog_test.go

@@ -34,10 +34,6 @@ func getFreeLocalhostPort() (int, error) {
 	return port, nil
 }
 
-func NoOpAck() error {
-	return nil
-}
-
 // Spawn a nats server and a JetStream instance for the duration of the test suite.
 // The different tests need to make sure to use different databases to avoid conflicts.
 var _ = SynchronizedBeforeSuite(func() {
@@ -149,7 +145,7 @@ var _ = Describe("ActivitylogService", func() {
 						}
 
 						for k, v := range tc.Activities {
-							err := alog.AddActivity(context.Background(), reference(v), nil, k, time.Time{}, getResource, NoOpAck)
+							err := alog.AddActivity(context.Background(), reference(v), nil, k, time.Time{}, getResource)
 							Expect(err).NotTo(HaveOccurred())
 						}
 					})
@@ -188,9 +184,9 @@ var _ = Describe("ActivitylogService", func() {
 
 			It("debounces activities", func() {
 
-				err := alog.AddActivity(context.Background(), reference("base"), nil, "activity1", time.Time{}, getResource, NoOpAck)
+				err := alog.AddActivity(context.Background(), reference("base"), nil, "activity1", time.Time{}, getResource)
 				Expect(err).NotTo(HaveOccurred())
-				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity2", time.Time{}, getResource, NoOpAck)
+				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity2", time.Time{}, getResource)
 				Expect(err).NotTo(HaveOccurred())
 
 				Eventually(func(g Gomega) {
@@ -201,7 +197,7 @@ var _ = Describe("ActivitylogService", func() {
 			})
 
 			It("adheres to the MaxActivities setting", func() {
-				err := alog.AddActivity(context.Background(), reference("base"), nil, "activity1", time.Time{}, getResource, NoOpAck)
+				err := alog.AddActivity(context.Background(), reference("base"), nil, "activity1", time.Time{}, getResource)
 				Expect(err).NotTo(HaveOccurred())
 				Eventually(func(g Gomega) {
 					activities, err := alog.Activities(resourceID("base"))
@@ -209,7 +205,7 @@ var _ = Describe("ActivitylogService", func() {
 					g.Expect(len(activities)).To(Equal(1))
 				}).Should(Succeed())
 
-				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity2", time.Time{}, getResource, NoOpAck)
+				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity2", time.Time{}, getResource)
 				Expect(err).NotTo(HaveOccurred())
 				Eventually(func(g Gomega) {
 					activities, err := alog.Activities(resourceID("base"))
@@ -217,11 +213,11 @@ var _ = Describe("ActivitylogService", func() {
 					g.Expect(len(activities)).To(Equal(2))
 				}).Should(Succeed())
 
-				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity3", time.Time{}, getResource, NoOpAck)
+				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity3", time.Time{}, getResource)
 				Expect(err).NotTo(HaveOccurred())
-				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity4", time.Time{}, getResource, NoOpAck)
+				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity4", time.Time{}, getResource)
 				Expect(err).NotTo(HaveOccurred())
-				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity5", time.Time{}, getResource, NoOpAck)
+				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity5", time.Time{}, getResource)
 				Expect(err).NotTo(HaveOccurred())
 
 				Eventually(func(g Gomega) {
@@ -238,9 +234,9 @@ var _ = Describe("ActivitylogService", func() {
 					return tree[ref.GetResourceId().GetOpaqueId()], nil
 				}
 
-				err := alog.AddActivity(context.Background(), reference("base"), nil, "activity1", time.Time{}, getResource, NoOpAck)
+				err := alog.AddActivity(context.Background(), reference("base"), nil, "activity1", time.Time{}, getResource)
 				Expect(err).NotTo(HaveOccurred())
-				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity2", time.Time{}, getResource, NoOpAck)
+				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity2", time.Time{}, getResource)
 				Expect(err).NotTo(HaveOccurred())
 
 				Eventually(func(g Gomega) {
@@ -249,9 +245,9 @@ var _ = Describe("ActivitylogService", func() {
 					g.Expect(activities).To(ConsistOf(activitites("activity1", 0, "activity2", 0)))
 				}).Should(Succeed())
 
-				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity3", time.Time{}, getResource, NoOpAck)
+				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity3", time.Time{}, getResource)
 				Expect(err).NotTo(HaveOccurred())
-				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity4", time.Time{}, getResource, NoOpAck)
+				err = alog.AddActivity(context.Background(), reference("base"), nil, "activity4", time.Time{}, getResource)
 				Expect(err).NotTo(HaveOccurred())
 
 				Eventually(func(g Gomega) {

services/activitylog/pkg/service/activitylog/debouncer.go

@@ -4,16 +4,15 @@ import (
 	"sync"
 	"time"
 
-	"github.com/opencloud-eu/opencloud/pkg/log"
 	"github.com/opencloud-eu/opencloud/services/activitylog/pkg/data"
 )
 
+// Debouncer is used to debounce writes to the activity log store.
 type Debouncer struct {
 	after      time.Duration
 	f          func(id string, ra []data.RawActivity) error
 	pending    sync.Map
 	inProgress sync.Map
-	log        log.Logger
 
 	mutex sync.Mutex
 }
@@ -23,19 +22,18 @@ type queueItem struct {
 	timer      *time.Timer
 }
 
-// NewDebouncer returns a new Debouncer instance
-func NewDebouncer(log log.Logger, d time.Duration, f func(id string, ra []data.RawActivity) error) *Debouncer {
+// NewDebouncer returns a new Debouncer instance.
+func NewDebouncer(d time.Duration, f func(id string, ra []data.RawActivity) error) *Debouncer {
 	return &Debouncer{
 		after:      d,
 		f:          f,
 		pending:    sync.Map{},
 		inProgress: sync.Map{},
-		log:        log,
 	}
 }
 
-// Debounce restarts the debounce timer for the given space
-func (d *Debouncer) Debounce(id string, ra data.RawActivity, ack func() error) {
+// Debounce restarts the debounce timer for the given space.
+func (d *Debouncer) Debounce(id string, ra data.RawActivity) {
 	if d.after == 0 {
 		d.f(id, []data.RawActivity{ra})
 		return
@@ -44,10 +42,10 @@ func (d *Debouncer) Debounce(id string, ra data.RawActivity, ack func() error) {
 	d.mutex.Lock()
 	defer d.mutex.Unlock()
 
-	activities := []data.RawActivity{ra}
 	item := &queueItem{
-		activities: activities,
+		activities: []data.RawActivity{ra},
 	}
+
 	if i, ok := d.pending.Load(id); ok {
 		// if the item is already in the queue, append the new activities
 		item, ok = i.(*queueItem)
@@ -73,13 +71,6 @@ func (d *Debouncer) Debounce(id string, ra data.RawActivity, ack func() error) {
 			d.inProgress.Store(id, true)
 			defer d.inProgress.Delete(id)
 			d.f(id, item.activities)
-			go func() {
-				if ack != nil {
-					if err := ack(); err != nil {
-						d.log.Error().Err(err).Msg("error while acknowledging event")
-					}
-				}
-			}()
 		})
 	}
 

services/activitylog/pkg/service/activitylog/debouncer_test.go

@@ -1,140 +1,68 @@
 package activitylog_test
 
 import (
-	"fmt"
 	"sync"
 	"time"
 
-	. "github.com/onsi/ginkgo/v2"
-	. "github.com/onsi/gomega"
-	"github.com/opencloud-eu/opencloud/pkg/log"
 	"github.com/opencloud-eu/opencloud/services/activitylog/pkg/data"
 	"github.com/opencloud-eu/opencloud/services/activitylog/pkg/service/activitylog"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
 )
 
-// ackTracker tracks per-name ack invocations so we can assert each event's ack fires.
-type ackTracker struct {
-	mu     sync.Mutex
-	calls  map[string]int
-	errors map[string]error
-}
-
-func newAckTracker() *ackTracker {
-	return &ackTracker{
-		calls:  make(map[string]int),
-		errors: make(map[string]error),
-	}
-}
-
-func (t *ackTracker) track(name string, fn func() error) func() error {
-	return func() error {
-		t.mu.Lock()
-		t.calls[name]++
-		err := fn()
-		t.errors[name] = err
-		t.mu.Unlock()
-		return err
-	}
-}
-
-func (t *ackTracker) count(name string) int {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	return t.calls[name]
-}
-
 var _ = Describe("Debouncer", func() {
 	var (
-		testlog   = log.NopLogger()
-		callbacks []data.RawActivity
-		mu        sync.Mutex
-		cbErr     error
-		ackCalled int
-		ackMu     sync.Mutex
+		mu         sync.Mutex
+		callbacks  []data.RawActivity
+		newCallback func(id string, ra []data.RawActivity) error
 	)
 
-	newCallback := func(_ string, ras []data.RawActivity) error {
-		mu.Lock()
-		defer mu.Unlock()
-		callbacks = append(callbacks, ras...)
-		return cbErr
-	}
-
-	newAck := func() error {
-		ackMu.Lock()
-		defer ackMu.Unlock()
-		ackCalled++
-		return nil
-	}
-
 	BeforeEach(func() {
+		mu.Lock()
 		callbacks = nil
-		cbErr = nil
-		ackMu.Lock()
-		ackCalled = 0
-		ackMu.Unlock()
+		mu.Unlock()
+		newCallback = func(id string, ra []data.RawActivity) error {
+			mu.Lock()
+			defer mu.Unlock()
+			callbacks = append(callbacks, ra...)
+			return nil
+		}
 	})
 
 	Context("with zero duration", func() {
-		It("calls the function immediately", func() {
-			d := activitylog.NewDebouncer(testlog, 0, newCallback)
-
-			ra := data.RawActivity{EventID: "immediate"}
-			d.Debounce("space1", ra, nil)
-
-			mu.Lock()
-			defer mu.Unlock()
+		It("calls the callback immediately", func() {
+			d := activitylog.NewDebouncer(0, newCallback)
+			d.Debounce("space1", data.RawActivity{EventID: "activity1"})
 			Expect(callbacks).To(HaveLen(1))
-			Expect(callbacks[0].EventID).To(Equal("immediate"))
+			Expect(callbacks[0].EventID).To(Equal("activity1"))
 		})
 
-		It("calls the function immediately for each call without batching", func() {
-			d := activitylog.NewDebouncer(testlog, 0, newCallback)
-
-			ra1 := data.RawActivity{EventID: "act1"}
-			ra2 := data.RawActivity{EventID: "act2"}
-			d.Debounce("space1", ra1, nil)
-			d.Debounce("space1", ra2, nil)
-
-			mu.Lock()
-			defer mu.Unlock()
+		It("calls the callback immediately for each event", func() {
+			d := activitylog.NewDebouncer(0, newCallback)
+			d.Debounce("space1", data.RawActivity{EventID: "activity1"})
+			d.Debounce("space2", data.RawActivity{EventID: "activity2"})
 			Expect(callbacks).To(HaveLen(2))
 		})
 	})
 
 	Context("with non-zero duration", func() {
 		It("batches activities with the same id", func() {
-			d := activitylog.NewDebouncer(testlog, 50*time.Millisecond, newCallback)
-
-			ra1 := data.RawActivity{EventID: "act1"}
-			ra2 := data.RawActivity{EventID: "act2"}
-			ra3 := data.RawActivity{EventID: "act3"}
-
-			d.Debounce("space1", ra1, nil)
-			d.Debounce("space1", ra2, nil)
-			d.Debounce("space1", ra3, nil)
+			d := activitylog.NewDebouncer(10*time.Millisecond, newCallback)
+			d.Debounce("space1", data.RawActivity{EventID: "activity1"})
+			d.Debounce("space1", data.RawActivity{EventID: "activity2"})
+			d.Debounce("space1", data.RawActivity{EventID: "activity3"})
 
 			Eventually(func() int {
 				mu.Lock()
 				defer mu.Unlock()
 				return len(callbacks)
 			}).Should(Equal(3))
-
-			mu.Lock()
-			defer mu.Unlock()
-			Expect(callbacks[0].EventID).To(Equal("act1"))
-			Expect(callbacks[1].EventID).To(Equal("act2"))
-			Expect(callbacks[2].EventID).To(Equal("act3"))
 		})
 
 		It("handles different ids independently", func() {
-			d := activitylog.NewDebouncer(testlog, 50*time.Millisecond, newCallback)
-
-			ra1 := data.RawActivity{EventID: "space1-act"}
-			ra2 := data.RawActivity{EventID: "space2-act"}
-
-			d.Debounce("space1", ra1, nil)
-			d.Debounce("space2", ra2, nil)
+			d := activitylog.NewDebouncer(10*time.Millisecond, newCallback)
+			d.Debounce("space1", data.RawActivity{EventID: "activity1"})
+			d.Debounce("space2", data.RawActivity{EventID: "activity2"})
 
 			Eventually(func() int {
 				mu.Lock()
@@ -143,58 +71,13 @@ var _ = Describe("Debouncer", func() {
 			}).Should(Equal(2))
 		})
 
-		It("calls the ack function after callback completes", func() {
-			d := activitylog.NewDebouncer(testlog, 50*time.Millisecond, newCallback)
-
-			ra := data.RawActivity{EventID: "act1"}
-			d.Debounce("space1", ra, newAck)
-
-			Eventually(func() int {
-				ackMu.Lock()
-				defer ackMu.Unlock()
-				return ackCalled
-			}).Should(Equal(1))
-		})
-
-		It("does not panic when ack is nil", func() {
-			d := activitylog.NewDebouncer(testlog, 50*time.Millisecond, newCallback)
-
-			ra := data.RawActivity{EventID: "act1"}
-			Expect(func() { d.Debounce("space1", ra, nil) }).ToNot(Panic())
-
-			Eventually(func() int {
-				mu.Lock()
-				defer mu.Unlock()
-				return len(callbacks)
-			}).Should(Equal(1))
-		})
-
-		It("does not panic when ack returns an error", func() {
-			d := activitylog.NewDebouncer(testlog, 50*time.Millisecond, newCallback)
-
-			errAck := func() error { return fmt.Errorf("ack error") }
-			ra := data.RawActivity{EventID: "act1"}
-			Expect(func() { d.Debounce("space1", ra, errAck) }).ToNot(Panic())
-
-			Eventually(func() int {
-				mu.Lock()
-				defer mu.Unlock()
-				return len(callbacks)
-			}).Should(Equal(1))
-		})
-
 		It("batches activities that arrive within the debounce window", func() {
-			d := activitylog.NewDebouncer(testlog, 200*time.Millisecond, newCallback)
-
-			ra1 := data.RawActivity{EventID: "act1"}
-			d.Debounce("space1", ra1, nil)
-
-			time.Sleep(50 * time.Millisecond)
-
-			ra2 := data.RawActivity{EventID: "act2"}
-			ra3 := data.RawActivity{EventID: "act3"}
-			d.Debounce("space1", ra2, nil)
-			d.Debounce("space1", ra3, nil)
+			d := activitylog.NewDebouncer(100*time.Millisecond, newCallback)
+			d.Debounce("space1", data.RawActivity{EventID: "activity1"})
+			time.Sleep(20 * time.Millisecond)
+			d.Debounce("space1", data.RawActivity{EventID: "activity2"})
+			time.Sleep(20 * time.Millisecond)
+			d.Debounce("space1", data.RawActivity{EventID: "activity3"})
 
 			Eventually(func() int {
 				mu.Lock()
@@ -204,235 +87,36 @@ var _ = Describe("Debouncer", func() {
 		})
 
 		It("processes new batch after previous completes", func() {
-			d := activitylog.NewDebouncer(testlog, 50*time.Millisecond, newCallback)
+			d := activitylog.NewDebouncer(5*time.Millisecond, newCallback)
+			d.Debounce("space1", data.RawActivity{EventID: "activity1"})
+			time.Sleep(20 * time.Millisecond) // let first batch complete
+			d.Debounce("space1", data.RawActivity{EventID: "activity2"})
 
-			// First batch
-			ra1 := data.RawActivity{EventID: "batch1-act1"}
-			ra2 := data.RawActivity{EventID: "batch1-act2"}
-			d.Debounce("space1", ra1, nil)
-			d.Debounce("space1", ra2, nil)
-
-			// Wait for first batch to be processed
 			Eventually(func() int {
 				mu.Lock()
 				defer mu.Unlock()
 				return len(callbacks)
 			}).Should(Equal(2))
-
-			// Second batch
-			callbacks = nil
-			ra3 := data.RawActivity{EventID: "batch2-act1"}
-			d.Debounce("space1", ra3, nil)
-
-			Eventually(func() int {
-				mu.Lock()
-				defer mu.Unlock()
-				return len(callbacks)
-			}).Should(Equal(1))
-			mu.Lock()
-			Expect(callbacks[0].EventID).To(Equal("batch2-act1"))
-			mu.Unlock()
 		})
 
-		It("reschedules when timer fires during in-progress callback", func() {
-			unblockCh := make(chan struct{})
-
-			blockingCallback := func(_ string, ras []data.RawActivity) error {
+		It("skips duplicate write when timer fires during in-progress callback", func() {
+			slowCallback := func(id string, ra []data.RawActivity) error {
+				time.Sleep(50 * time.Millisecond) // simulate slow write
 				mu.Lock()
-				callbacks = append(callbacks, ras...)
-				mu.Unlock()
-
-				<-unblockCh
+				defer mu.Unlock()
+				callbacks = append(callbacks, ra...)
 				return nil
 			}
 
-			d := activitylog.NewDebouncer(testlog, 50*time.Millisecond, blockingCallback)
-
-			// First activity starts the timer
-			ra1 := data.RawActivity{EventID: "act1"}
-			d.Debounce("space1", ra1, nil)
-
-			// Second activity gets queued while first is being processed
-			time.Sleep(60 * time.Millisecond)
-			// Timer fires for act1, callback starts (and blocks)
-			Eventually(func() int {
-				mu.Lock()
-				defer mu.Unlock()
-				return len(callbacks)
-			}, 1*time.Second).Should(Equal(1))
-
-			// Queue another activity while the first callback is still running
-			ra2 := data.RawActivity{EventID: "act2"}
-			d.Debounce("space1", ra2, nil)
-
-			// Unblock the first callback
-			close(unblockCh)
-
-			// The second activity should eventually be processed
-			Eventually(func() int {
-				mu.Lock()
-				defer mu.Unlock()
-				return len(callbacks)
-			}, 2*time.Second).Should(Equal(2))
-
-			mu.Lock()
-			defer mu.Unlock()
-			Expect(callbacks[0].EventID).To(Equal("act1"))
-			Expect(callbacks[1].EventID).To(Equal("act2"))
-		})
-
-		It("calls ack even when callback returns an error", func() {
-			cbErr = fmt.Errorf("store failed")
-			d := activitylog.NewDebouncer(testlog, 50*time.Millisecond, newCallback)
-
-			ra := data.RawActivity{EventID: "act1"}
-			d.Debounce("space1", ra, newAck)
-
-			Eventually(func() int {
-				ackMu.Lock()
-				defer ackMu.Unlock()
-				return ackCalled
-			}).Should(Equal(1))
-		})
-
-		It("calls each ack when multiple events target the same id within one batch", func() {
-			tracker := newAckTracker()
-			d := activitylog.NewDebouncer(testlog, 200*time.Millisecond, newCallback)
-
-			ra1 := data.RawActivity{EventID: "act1"}
-			ra2 := data.RawActivity{EventID: "act2"}
-			ra3 := data.RawActivity{EventID: "act3"}
-
-			d.Debounce("space1", ra1, tracker.track("ack1", func() error { return nil }))
-			d.Debounce("space1", ra2, tracker.track("ack2", func() error { return nil }))
-			d.Debounce("space1", ra3, tracker.track("ack3", func() error { return nil }))
-
-			// All three activities should be stored ...
-			Eventually(func() int {
-				mu.Lock()
-				defer mu.Unlock()
-				return len(callbacks)
-			}).Should(Equal(3))
-
-			// ... and all three acks should have been called.
-			Eventually(func() int { return tracker.count("ack1") }).Should(Equal(1))
-			Eventually(func() int { return tracker.count("ack2") }).Should(Equal(1))
-			Eventually(func() int { return tracker.count("ack3") }).Should(Equal(1))
-		})
-
-		It("calls ack on the re-schedule path when timer fires during in-progress callback", func() {
-			tracker := newAckTracker()
-			unblockCh := make(chan struct{})
-
-			blockingCallback := func(_ string, ras []data.RawActivity) error {
-				mu.Lock()
-				callbacks = append(callbacks, ras...)
-				mu.Unlock()
-
-				<-unblockCh
-				return nil
-			}
-
-			d := activitylog.NewDebouncer(testlog, 50*time.Millisecond, blockingCallback)
-
-			// First activity starts the timer.
-			ra1 := data.RawActivity{EventID: "act1"}
-			d.Debounce("space1", ra1, tracker.track("ack1", func() error { return nil }))
-
-			// Wait for timer to fire so the blocking callback begins.
-			Eventually(func() int {
-				mu.Lock()
-				defer mu.Unlock()
-				return len(callbacks)
-			}, 1*time.Second).Should(Equal(1))
-
-			// Queue a second activity on a *different* id so its own timer fires
-			// while the first callback is still blocked. When that timer fires,
-			// it will complete immediately (no blocking), then we'll queue onto
-			// space1 again to hit the re-schedule path.
-			ra2 := data.RawActivity{EventID: "act2"}
-			d.Debounce("space2", ra2, tracker.track("ack2", func() error { return nil }))
-
-			// Wait for space2 batch to complete (non-blocking callback).
-			Eventually(func() int {
-				mu.Lock()
-				defer mu.Unlock()
-				return len(callbacks)
-			}, 1*time.Second).Should(Equal(2))
-
-			// Now queue onto space1 while its callback is still in-progress.
-			// The timer for space1 has already fired once; new Debounce calls
-			// will append to the existing pending item and reset the timer,
-			// which triggers the re-schedule path when it fires again.
-			ra3 := data.RawActivity{EventID: "act3"}
-			d.Debounce("space1", ra3, tracker.track("ack3", func() error { return nil }))
-
-			// Unblock the first callback so space1 can process act3.
-			close(unblockCh)
-
-			Eventually(func() int {
-				mu.Lock()
-				defer mu.Unlock()
-				return len(callbacks)
-			}, 2*time.Second).Should(Equal(3))
-
-			// All three acks should have been called.
-			Eventually(func() int { return tracker.count("ack1") }).Should(Equal(1))
-			Eventually(func() int { return tracker.count("ack2") }).Should(Equal(1))
-			Eventually(func() int { return tracker.count("ack3") }).Should(Equal(1))
-		})
-
-		It("acks event even when store callback fails (data-loss risk)", func() {
-			// This test documents current behaviour: ack fires regardless of
-			// whether StoreActivity succeeded.  If storage fails but the event
-			// is still acked, it will never be replayed by JetStream.
-			cbErr = fmt.Errorf("nats kv put failed")
-			d := activitylog.NewDebouncer(testlog, 50*time.Millisecond, newCallback)
-
-			ra := data.RawActivity{EventID: "act1"}
-			d.Debounce("space1", ra, newAck)
+			d := activitylog.NewDebouncer(10*time.Millisecond, slowCallback)
+			d.Debounce("space1", data.RawActivity{EventID: "activity1"})
+			time.Sleep(20 * time.Millisecond) // timer fires while callback is running
 
 			Eventually(func() int {
 				mu.Lock()
 				defer mu.Unlock()
 				return len(callbacks)
 			}).Should(Equal(1))
-
-			// ack was called despite the store error -- event is lost.
-			Eventually(func() int {
-				ackMu.Lock()
-				defer ackMu.Unlock()
-				return ackCalled
-			}).Should(Equal(1))
-		})
-	})
-
-	Context("zero-duration ack behaviour", func() {
-		It("calls ack when duration is zero (immediate path)", func() {
-			d := activitylog.NewDebouncer(testlog, 0, newCallback)
-
-			ra := data.RawActivity{EventID: "immediate"}
-			d.Debounce("space1", ra, newAck)
-
-			ackMu.Lock()
-			defer ackMu.Unlock()
-			Expect(ackCalled).To(Equal(1))
-		})
-
-		It("calls each ack when multiple immediate events target the same id", func() {
-			tracker := newAckTracker()
-			d := activitylog.NewDebouncer(testlog, 0, newCallback)
-
-			ra1 := data.RawActivity{EventID: "act1"}
-			ra2 := data.RawActivity{EventID: "act2"}
-
-			d.Debounce("space1", ra1, tracker.track("ack1", func() error { return nil }))
-			d.Debounce("space1", ra2, tracker.track("ack2", func() error { return nil }))
-
-			Expect(tracker.count("ack1")).To(Equal(1))
-			Expect(tracker.count("ack2")).To(Equal(1))
 		})
 	})
 })
-
-

services/activitylog/pkg/service/events/service.go

@@ -15,7 +15,6 @@ import (
 	"github.com/opencloud-eu/opencloud/services/activitylog/pkg/data"
 	"github.com/opencloud-eu/opencloud/services/activitylog/pkg/service/activitylog"
 	"github.com/opencloud-eu/reva/v2/pkg/events"
-	"github.com/opencloud-eu/reva/v2/pkg/events/raw"
 	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
 	"github.com/opencloud-eu/reva/v2/pkg/utils"
@@ -38,7 +37,7 @@ type ActivitylogService struct {
 	ctx    context.Context
 	sa     config.ServiceAccount
 	log    log.Logger
-	stream raw.Stream
+	stream events.Stream
 	gws    pool.Selectable[gateway.GatewayAPIClient]
 	al     *activitylog.ActivityLog
 
@@ -51,7 +50,7 @@ type ActivitylogService struct {
 }
 
 // New creates a new ActivitylogService
-func New(al *activitylog.ActivityLog, stream raw.Stream, opts ...Option) (*ActivitylogService, error) {
+func New(al *activitylog.ActivityLog, stream events.Stream, opts ...Option) (*ActivitylogService, error) {
 	o := &Options{
 		NumConsumers: _numConsumersDefault,
 	}
@@ -77,15 +76,11 @@ func New(al *activitylog.ActivityLog, stream raw.Stream, opts ...Option) (*Activ
 
 // Run to fulfil Runner interface
 func (s *ActivitylogService) Run() error {
-	ch, err := s.stream.Consume("activitylog-pull", s.events...)
+	ch, err := events.Consume(s.stream, "activitylog-pull", s.events...)
 	if err != nil {
 		return err
 	}
 
-	// if s.m != nil {
-	// 	monitorMetrics(s.ctx, s.stream, "activitylog", s.m, s.log)
-	// }
-
 	var wg sync.WaitGroup
 	ctx, cancel := context.WithCancel(s.ctx)
 	defer cancel()
@@ -130,30 +125,24 @@ func (s *ActivitylogService) Run() error {
 
 // Close will make the service to stop processing, so the `Run`
 // method can finish.
-// TODO: Underlying services can't be stopped. This means that some goroutines
-// will get stuck trying to push events through a channel nobody is reading
-// from, so resources won't be freed and there will be memory leaks. For now,
-// if the service is stopped, you should close the app soon after.
 func (s *ActivitylogService) Close() {
 	if s.stopped.CompareAndSwap(false, true) {
 		close(s.stopCh)
 	}
 }
 
-// Run runs the service
-func (s *ActivitylogService) processEvent(e raw.Event) error {
+func (s *ActivitylogService) processEvent(e events.Event) error {
 	ctx := e.GetTraceContext(s.ctx)
 	ctx, span := tracer.Start(ctx, "processEvent")
 	defer span.End()
 
-	e.InProgress() // let nats know that we are processing this event
 	s.log.Debug().Interface("event", e).Msg("updating activitylog")
 
-	switch ev := e.Event.Event.(type) {
+	switch ev := e.Event.(type) {
 	case events.UploadReady:
-		return s.AddActivity(ctx, ev.FileRef, ev.ParentID, e.ID, utils.TSToTime(ev.Timestamp), e.Ack)
+		return s.AddActivity(ctx, ev.FileRef, ev.ParentID, e.ID, utils.TSToTime(ev.Timestamp))
 	case events.FileTouched:
-		return s.AddActivity(ctx, ev.Ref, ev.ParentID, e.ID, utils.TSToTime(ev.Timestamp), e.Ack)
+		return s.AddActivity(ctx, ev.Ref, ev.ParentID, e.ID, utils.TSToTime(ev.Timestamp))
 	// Disabled https://github.com/owncloud/ocis/issues/10293
 	//case events.FileDownloaded:
 	// we are only interested in public link downloads - so no need to store others.
@@ -161,57 +150,43 @@ func (s *ActivitylogService) processEvent(e raw.Event) error {
 	//	err = a.AddActivity(ev.Ref, e.ID, utils.TSToTime(ev.Timestamp))
 	//}
 	case events.ContainerCreated:
-		return s.AddActivity(ctx, ev.Ref, ev.ParentID, e.ID, utils.TSToTime(ev.Timestamp), e.Ack)
+		return s.AddActivity(ctx, ev.Ref, ev.ParentID, e.ID, utils.TSToTime(ev.Timestamp))
 	case events.ItemTrashed:
-		return s.AddActivityTrashed(ctx, ev.ID, ev.Ref, nil, e.ID, utils.TSToTime(ev.Timestamp), e.Ack)
+		return s.AddActivityTrashed(ctx, ev.ID, ev.Ref, nil, e.ID, utils.TSToTime(ev.Timestamp))
 	case events.ItemPurged:
-		return s.al.RemoveResource(ev.ID, e.Ack)
+		return s.al.RemoveResource(ev.ID)
 	case events.ItemMoved:
 		// remove the cached parent id for this resource
 		s.removeCachedParentID(ctx, ev.Ref)
 
-		return s.AddActivity(ctx, ev.Ref, nil, e.ID, utils.TSToTime(ev.Timestamp), e.Ack)
+		return s.AddActivity(ctx, ev.Ref, nil, e.ID, utils.TSToTime(ev.Timestamp))
 	case events.ShareCreated:
-		return s.AddActivity(ctx, toRef(ev.ItemID), nil, e.ID, utils.TSToTime(ev.CTime), e.Ack)
+		return s.AddActivity(ctx, toRef(ev.ItemID), nil, e.ID, utils.TSToTime(ev.CTime))
 	case events.ShareUpdated:
 		if ev.Sharer != nil && ev.ItemID != nil && ev.Sharer.GetOpaqueId() != ev.ItemID.GetSpaceId() {
-			return s.AddActivity(ctx, toRef(ev.ItemID), nil, e.ID, utils.TSToTime(ev.MTime), e.Ack)
-		} else {
-			go func() {
-				err := e.Ack()
-				if err != nil {
-					s.log.Error().Err(err).Msg("error while acknowledging event")
-				}
-			}()
+			return s.AddActivity(ctx, toRef(ev.ItemID), nil, e.ID, utils.TSToTime(ev.MTime))
 		}
 	case events.ShareRemoved:
-		return s.AddActivity(ctx, toRef(ev.ItemID), nil, e.ID, ev.Timestamp, e.Ack)
+		return s.AddActivity(ctx, toRef(ev.ItemID), nil, e.ID, ev.Timestamp)
 	case events.LinkCreated:
-		return s.AddActivity(ctx, toRef(ev.ItemID), nil, e.ID, utils.TSToTime(ev.CTime), e.Ack)
+		return s.AddActivity(ctx, toRef(ev.ItemID), nil, e.ID, utils.TSToTime(ev.CTime))
 	case events.LinkUpdated:
 		if ev.Sharer != nil && ev.ItemID != nil && ev.Sharer.GetOpaqueId() != ev.ItemID.GetSpaceId() {
-			return s.AddActivity(ctx, toRef(ev.ItemID), nil, e.ID, utils.TSToTime(ev.MTime), e.Ack)
-		} else {
-			go func() {
-				err := e.Ack()
-				if err != nil {
-					s.log.Error().Err(err).Msg("error while acknowledging event")
-				}
-			}()
+			return s.AddActivity(ctx, toRef(ev.ItemID), nil, e.ID, utils.TSToTime(ev.MTime))
 		}
 	case events.LinkRemoved:
-		return s.AddActivity(ctx, toRef(ev.ItemID), nil, e.ID, utils.TSToTime(ev.Timestamp), e.Ack)
+		return s.AddActivity(ctx, toRef(ev.ItemID), nil, e.ID, utils.TSToTime(ev.Timestamp))
 	case events.SpaceShared:
-		return s.AddSpaceActivity(ctx, ev.ID, e.ID, ev.Timestamp, e.Ack)
+		return s.AddSpaceActivity(ctx, ev.ID, e.ID, ev.Timestamp)
 	case events.SpaceUnshared:
-		return s.AddSpaceActivity(ctx, ev.ID, e.ID, ev.Timestamp, e.Ack)
+		return s.AddSpaceActivity(ctx, ev.ID, e.ID, ev.Timestamp)
 	}
 
 	return nil
 }
 
 // AddActivity adds the activity to the given resource and all its parents
-func (a *ActivitylogService) AddActivity(ctx context.Context, initRef *provider.Reference, parentId *provider.ResourceId, eventID string, timestamp time.Time, ack func() error) error {
+func (a *ActivitylogService) AddActivity(ctx context.Context, initRef *provider.Reference, parentId *provider.ResourceId, eventID string, timestamp time.Time) error {
 	ctx, span := tracer.Start(ctx, "AddActivity")
 	defer span.End()
 
@@ -227,11 +202,11 @@ func (a *ActivitylogService) AddActivity(ctx context.Context, initRef *provider.
 	}
 	return a.al.AddActivity(ctx, initRef, parentId, eventID, timestamp, func(ctx context.Context, ref *provider.Reference) (*provider.ResourceInfo, error) {
 		return utils.GetResource(ctx, ref, gwc)
-	}, ack)
+	})
 }
 
 // AddActivityTrashed adds the activity to given trashed resource and all its former parents
-func (a *ActivitylogService) AddActivityTrashed(ctx context.Context, resourceID *provider.ResourceId, reference *provider.Reference, parentId *provider.ResourceId, eventID string, timestamp time.Time, ack func() error) error {
+func (a *ActivitylogService) AddActivityTrashed(ctx context.Context, resourceID *provider.ResourceId, reference *provider.Reference, parentId *provider.ResourceId, eventID string, timestamp time.Time) error {
 	ctx, span := tracer.Start(ctx, "AddActivityTrashed")
 	defer span.End()
 
@@ -264,11 +239,11 @@ func (a *ActivitylogService) AddActivityTrashed(ctx context.Context, resourceID
 
 	return a.al.AddActivity(ctx, ref, parentId, eventID, timestamp, func(ctx context.Context, ref *provider.Reference) (*provider.ResourceInfo, error) {
 		return utils.GetResource(ctx, ref, gwc)
-	}, ack)
+	})
 }
 
 // AddSpaceActivity adds the activity to the given spaceroot
-func (a *ActivitylogService) AddSpaceActivity(ctx context.Context, spaceID *provider.StorageSpaceId, eventID string, timestamp time.Time, ack func() error) error {
+func (a *ActivitylogService) AddSpaceActivity(ctx context.Context, spaceID *provider.StorageSpaceId, eventID string, timestamp time.Time) error {
 	_, span := tracer.Start(ctx, "AddSpaceActivity")
 	defer span.End()
 	// spaceID is in format <providerid>$<spaceid>
@@ -289,12 +264,6 @@ func (a *ActivitylogService) AddSpaceActivity(ctx context.Context, spaceID *prov
 	if err != nil {
 		return fmt.Errorf("could not store activity: %w", err)
 	}
-	go func() {
-		err := ack()
-		if err != nil {
-			a.log.Error().Err(err).Msg("error while acknowledging event")
-		}
-	}()
 	return nil
 }