Make sure all timestamps are UTC (#40)

Author: raphael

pool/node.go

@@ -541,7 +541,7 @@ func (node *Node) routeWorkerEvent(ctx context.Context, ev *streaming.Event) err
 	node.logger.Debug("routed", "event", ev.EventName, "id", ev.ID, "worker", wid, "worker-event-id", eventID)
 
 	// Record the event in the pending events map for future ack.
-	node.pendingEvents[wid+":"+eventID] = ev
+	node.pendingEvents[pendingEventKey(wid, eventID)] = ev
 
 	return nil
 }
@@ -583,7 +583,7 @@ func (node *Node) ackWorkerEvent(ctx context.Context, ev *streaming.Event) {
 
 	workerID, payload := unmarshalEnvelope(ev.Payload)
 	ack := unmarshalAck(payload)
-	key := workerID + ":" + ack.EventID
+	key := pendingEventKey(workerID, ack.EventID)
 	pending, ok := node.pendingEvents[key]
 	if !ok {
 		node.logger.Error(fmt.Errorf("ackWorkerEvent: received unknown event %s from worker %s", ack.EventID, workerID))
@@ -619,7 +619,7 @@ func (node *Node) ackWorkerEvent(ctx context.Context, ev *streaming.Event) {
 		}
 	}
 	for _, key := range staleKeys {
-		node.logger.Error(fmt.Errorf("ackWorkerEvent: stale event, removing from pending events"), "event", node.pendingEvents[key].EventName, "id", key)
+		node.logger.Error(fmt.Errorf("ackWorkerEvent: stale event, removing from pending events"), "event", node.pendingEvents[key].EventName, "id", node.pendingEvents[key].ID, "since", time.Since(node.pendingEvents[key].CreatedAt()), "TTL", 2*node.pendingJobTTL)
 		delete(node.pendingEvents, key)
 	}
 }
@@ -1028,3 +1028,9 @@ func poolStreamName(pool string) string {
 func nodeStreamName(pool, nodeID string) string {
 	return fmt.Sprintf("%s:node:%s", pool, nodeID)
 }
+
+// pendingEventKey computes the key of a pending event from a worker ID and a
+// stream event ID.
+func pendingEventKey(workerID, eventID string) string {
+	return fmt.Sprintf("%s:%s", workerID, eventID)
+}

pool/node_test.go

@@ -519,7 +519,7 @@ func TestStaleEventsAreRemoved(t *testing.T) {
 			},
 		},
 	}
-	node.pendingEvents["worker:stale-event-id"] = staleEvent
+	node.pendingEvents[pendingEventKey("worker", staleEventID)] = staleEvent
 
 	// Add a fresh event
 	freshEventID := fmt.Sprintf("%d-0", time.Now().Add(-time.Second).UnixNano()/int64(time.Millisecond))
@@ -533,35 +533,36 @@ func TestStaleEventsAreRemoved(t *testing.T) {
 			},
 		},
 	}
-	node.pendingEvents["worker:fresh-event-id"] = freshEvent
+	node.pendingEvents[pendingEventKey("worker", freshEventID)] = freshEvent
 
 	// Create a mock event to trigger the ackWorkerEvent function
+	mockEventID := "mock-event-id"
 	mockEvent := &streaming.Event{
-		ID:        "mock-event-id",
+		ID:        mockEventID,
 		EventName: evAck,
-		Payload:   marshalEnvelope("worker", marshalAck(&ack{EventID: "mock-event-id"})),
+		Payload:   marshalEnvelope("worker", marshalAck(&ack{EventID: mockEventID})),
 		Acker: &mockAcker{
 			XAckFunc: func(ctx context.Context, streamKey, sinkName string, ids ...string) *redis.IntCmd {
 				return redis.NewIntCmd(ctx, 0)
 			},
 		},
 	}
-	node.pendingEvents["worker:mock-event-id"] = mockEvent
+	node.pendingEvents[pendingEventKey("worker", mockEventID)] = mockEvent
 
 	// Call ackWorkerEvent to trigger the stale event cleanup
 	node.ackWorkerEvent(ctx, mockEvent)
 
 	assert.Eventually(t, func() bool {
 		node.lock.Lock()
 		defer node.lock.Unlock()
-		_, exists := node.pendingEvents["worker:stale-event-id"]
+		_, exists := node.pendingEvents[pendingEventKey("worker", staleEventID)]
 		return !exists
 	}, max, delay, "Stale event should have been removed")
 
 	assert.Eventually(t, func() bool {
 		node.lock.Lock()
 		defer node.lock.Unlock()
-		_, exists := node.pendingEvents["worker:fresh-event-id"]
+		_, exists := node.pendingEvents[pendingEventKey("worker", freshEventID)]
 		return exists
 	}, max, delay, "Fresh event should still be present")
 }

streaming/reader.go

@@ -288,7 +288,7 @@ func (e *Event) CreatedAt() time.Time {
 	ts, _ := strconv.ParseInt(tss, 10, 64)
 	seconds := ts / 1000
 	nanos := (ts % 1000) * 1_000_000
-	return time.Unix(seconds, nanos)
+	return time.Unix(seconds, nanos).UTC()
 }
 
 // streamEvents filters and streams the Redis messages as events to c.

streaming/reader_test.go

@@ -1,12 +1,15 @@
 package streaming
 
 import (
+	"strconv"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/redis/go-redis/v9"
 	"goa.design/pulse/pulse"
 	"goa.design/pulse/streaming/options"
 	ptesting "goa.design/pulse/testing"
@@ -185,3 +188,33 @@ func TestRemoveReaderStream(t *testing.T) {
 	assert.Equal(t, "event3", read.EventName)
 	assert.Equal(t, []byte("payload3"), read.Payload)
 }
+
+func TestEventCreatedAt(t *testing.T) {
+	rdb := ptesting.NewRedisClient(t)
+	defer ptesting.CleanupRedis(t, rdb, false, "")
+	ctx := ptesting.NewTestContext(t)
+
+	// Use Redis to create a new event ID
+	eventID, err := rdb.XAdd(ctx, &redis.XAddArgs{
+		Stream: "test-stream",
+		Values: map[string]interface{}{"key": "value"},
+	}).Result()
+	require.NoError(t, err)
+
+	event := &Event{ID: eventID}
+
+	// Call CreatedAt() method
+	createdAt := event.CreatedAt()
+
+	// Parse the timestamp from the event ID
+	parts := strings.Split(eventID, "-")
+	ts, err := strconv.ParseInt(parts[0], 10, 64)
+	require.NoError(t, err)
+	expectedTime := time.UnixMilli(ts).UTC()
+
+	// Assert that the returned time is exactly the expected time
+	assert.Equal(t, expectedTime, createdAt)
+
+	// Assert that the returned time is in UTC
+	assert.Equal(t, time.UTC, createdAt.Location())
+}