Merge pull request #8 from xaionaro-go/drafts

[youtube] Investigating quota consumption mismatch

Author: xaionaro

pkg/streamcontrol/youtube/chat_listener.go

@@ -211,8 +211,11 @@ func (l *ChatListener) streamMessages(
 	}
 	logger.Debugf(ctx, "gRPC stream opened successfully for liveChatId=%q", l.liveChatID)
 
+	// StreamList quota cost is not documented by Google (Live Streaming API
+	// operations are absent from the quota calculator). Set to 5 as a
+	// provisional estimate; needs further empirical calibration.
 	if l.quotaTracker != nil {
-		l.quotaTracker.ReportQuotaConsumption(ctx, "StreamList", 1)
+		l.quotaTracker.ReportQuotaConsumption(ctx, "StreamList", 5)
 	}
 
 	streamStartedAt := time.Now()

pkg/streamcontrol/youtube/chat_listener_test.go

@@ -612,8 +612,8 @@ func TestChatListenerQuotaTracking(t *testing.T) {
 	err = listener.streamMessages(ctx, client, &pageToken)
 	require.NoError(t, err)
 
-	assert.Equal(t, uint64(1), qt.UsedQuotaPoints(),
-		"each StreamList RPC should report 1 quota point (matching api/request_count)")
+	assert.Equal(t, uint64(5), qt.UsedQuotaPoints(),
+		"StreamList quota is charged per gRPC stream open (provisional estimate: 5 units)")
 	assert.Equal(t, fmt.Sprintf("token-%d", responseCount-1), pageToken)
 	assert.Equal(t, responseCount, len(listener.messagesOutChan),
 		"all messages should be delivered to the output channel")

pkg/streamcontrol/youtube/types/config.go

@@ -26,7 +26,6 @@ type AccountConfig struct {
 	ClientSecret         secret.String
 	Token                *OAuth2Token
 	GCPProjectID         string          `yaml:"GCPProjectID,omitempty"`
-	QuotaUsedPoints      uint64            `yaml:"quota_used_points,omitempty"`
 	QuotaUsedByOp        map[string]uint64 `yaml:"quota_used_by_op,omitempty"`
 	QuotaUsedDate        string            `yaml:"quota_used_date,omitempty"`
 	CustomOAuthHandler   OAuthHandler    `yaml:"-"`

pkg/streamcontrol/youtube/youtube.go

@@ -133,13 +133,15 @@ func New(
 		return nil, fmt.Errorf("initialization failed: %w", err)
 	}
 
-	if cfg.QuotaUsedPoints > 0 && cfg.QuotaUsedDate == getQuotaCutoffDate(time.Now()) {
-		yt.YouTubeClient.UsedPoints.Store(cfg.QuotaUsedPoints)
+	if len(cfg.QuotaUsedByOp) > 0 && cfg.QuotaUsedDate == getQuotaCutoffDate(time.Now()) {
+		var total uint64
 		for op, pts := range cfg.QuotaUsedByOp {
 			yt.YouTubeClient.UsedPointsByOp.Store(op, pts)
+			total += pts
 		}
+		yt.YouTubeClient.UsedPoints.Store(total)
 		yt.YouTubeClient.PreviousCheckAt = time.Now()
-		logger.Infof(ctx, "loaded persisted quota: %d points for %s", cfg.QuotaUsedPoints, cfg.QuotaUsedDate)
+		logger.Infof(ctx, "loaded persisted quota: %d points for %s", total, cfg.QuotaUsedDate)
 	}
 
 	err = yt.YouTubeClient.Ping(ctx)
@@ -212,20 +214,20 @@ func (yt *YouTube) persistQuota(ctx context.Context) {
 		return
 	}
 
-	points := yt.YouTubeClient.UsedPoints.Load()
 	date := getQuotaCutoffDate(time.Now())
 
 	byOp := map[string]uint64{}
+	var total uint64
 	yt.YouTubeClient.UsedPointsByOp.Range(func(key string, value uint64) bool {
 		byOp[key] = value
+		total += value
 		return true
 	})
 
-	if yt.Config.QuotaUsedPoints == points && yt.Config.QuotaUsedDate == date {
+	if sumQuotaByOp(yt.Config.QuotaUsedByOp) == total && yt.Config.QuotaUsedDate == date {
 		return
 	}
 
-	yt.Config.QuotaUsedPoints = points
 	yt.Config.QuotaUsedByOp = byOp
 	yt.Config.QuotaUsedDate = date
 
@@ -234,6 +236,14 @@ func (yt *YouTube) persistQuota(ctx context.Context) {
 	}
 }
 
+func sumQuotaByOp(m map[string]uint64) uint64 {
+	var total uint64
+	for _, v := range m {
+		total += v
+	}
+	return total
+}
+
 func (yt *YouTube) getNewToken(ctx context.Context) (_ret *oauth2.Token, _err error) {
 	logger.Debugf(ctx, "YouTube.getNewToken")
 	defer func() { logger.Debugf(ctx, "/YouTube.getNewToken: %v", _err) }()
@@ -1454,7 +1464,7 @@ func (yt *YouTube) GetStreamStatus(
 		return nil, fmt.Errorf("unable to get active broadcasts info: %w", err)
 	}
 	if len(requestStatsVideoIDs) > 0 {
-		videos, err := yt.YouTubeClient.Client.GetVideos(ctx, requestStatsVideoIDs, videoParts)
+		videos, err := yt.YouTubeClient.GetVideos(ctx, requestStatsVideoIDs, videoParts)
 		if err != nil {
 			logger.Errorf(ctx, "unable to get info for videos %v: %v", requestStatsVideoIDs, err)
 		} else {

pkg/streamcontrol/youtube/youtube_test.go

@@ -159,7 +159,8 @@ func TestGetInfoReturnsQuotaAndListeners(t *testing.T) {
 	assert.Empty(t, info.ChatListeners)
 	assert.Empty(t, info.ActiveBroadcasts)
 
-	// Simulate some quota usage.
+	// Simulate some quota usage via per-op tracking.
+	yt.YouTubeClient.UsedPointsByOp.Store("GetBroadcasts", 42)
 	yt.YouTubeClient.UsedPoints.Store(42)
 	info = yt.GetInfo(ctx)
 	assert.Equal(t, uint64(42), info.QuotaUsage.UsedPoints.Load())
@@ -215,18 +216,20 @@ func TestTemplateBroadcastIDSet(t *testing.T) {
 
 func TestQuotaPersistenceLoad(t *testing.T) {
 	cfg := newTestAccountConfig()
-	cfg.QuotaUsedPoints = 500
+	cfg.QuotaUsedByOp = map[string]uint64{
+		"GetBroadcasts": 200,
+		"Search":        300,
+	}
 	cfg.QuotaUsedDate = getQuotaCutoffDate(time.Now())
 
 	yt := newTestYouTubeWithConfig(t, cfg, func(AccountConfig) error { return nil })
 
-	// 500 loaded + 1 from Ping during New()
+	// 500 loaded from per-op map + 1 from Ping during New()
 	assert.Equal(t, uint64(501), yt.YouTubeClient.UsedPoints.Load())
 }
 
 func TestQuotaPersistenceLoadPerOp(t *testing.T) {
 	cfg := newTestAccountConfig()
-	cfg.QuotaUsedPoints = 200
 	cfg.QuotaUsedDate = getQuotaCutoffDate(time.Now())
 	cfg.QuotaUsedByOp = map[string]uint64{
 		"GetBroadcasts": 3,
@@ -248,11 +251,17 @@ func TestQuotaPersistenceLoadPerOp(t *testing.T) {
 	updVid, ok := yt.YouTubeClient.UsedPointsByOp.Load("UpdateVideo")
 	assert.True(t, ok)
 	assert.Equal(t, uint64(97), updVid)
+
+	// Total should be derived from per-op sum + 1 from Ping during New()
+	assert.Equal(t, uint64(201), yt.YouTubeClient.UsedPoints.Load())
 }
 
 func TestQuotaPersistenceLoadStaleDate(t *testing.T) {
 	cfg := newTestAccountConfig()
-	cfg.QuotaUsedPoints = 500
+	cfg.QuotaUsedByOp = map[string]uint64{
+		"GetBroadcasts": 200,
+		"Search":        300,
+	}
 	cfg.QuotaUsedDate = "2020-01-01"
 
 	yt := newTestYouTubeWithConfig(t, cfg, func(AccountConfig) error { return nil })
@@ -276,32 +285,22 @@ func TestPersistQuotaSavesOnChange(t *testing.T) {
 	yt := newTestYouTubeWithConfig(t, cfg, saveFn)
 	ctx := context.Background()
 
-	// Simulate some quota usage.
-	yt.YouTubeClient.UsedPoints.Store(42)
-
-	// First persist should save.
+	// After New(), Ping added 1 point. First explicit persist should save.
 	saveCount = 0
 	yt.persistQuota(ctx)
 	assert.Equal(t, 1, saveCount)
-	assert.Equal(t, uint64(42), savedCfg.QuotaUsedPoints)
+	assert.Equal(t, map[string]uint64{"Ping": 1}, savedCfg.QuotaUsedByOp)
 	assert.Equal(t, getQuotaCutoffDate(time.Now()), savedCfg.QuotaUsedDate)
 
 	// Second persist with same value should not save again.
 	yt.persistQuota(ctx)
 	assert.Equal(t, 1, saveCount, "persistQuota should skip save when values unchanged")
 
-	// Change the value; persist should save again.
-	yt.YouTubeClient.UsedPoints.Store(100)
-	yt.persistQuota(ctx)
-	assert.Equal(t, 2, saveCount)
-	assert.Equal(t, uint64(100), savedCfg.QuotaUsedPoints)
-
-	// Verify per-op breakdown is persisted.
+	// Add more per-op usage; persist should save again.
 	yt.YouTubeClient.UsedPointsByOp.Store("GetBroadcasts", 5)
 	yt.YouTubeClient.UsedPointsByOp.Store("Search", 95)
-	yt.YouTubeClient.UsedPoints.Store(200)
 	yt.persistQuota(ctx)
-	assert.Equal(t, 3, saveCount)
+	assert.Equal(t, 2, saveCount)
 	assert.Equal(t, map[string]uint64{
 		"GetBroadcasts": 5,
 		"Ping":          1,