Method FlushToDB

coderd/boundaryusage/tracker.go:73–134

FlushToDB writes stats to the database. For unique counts, cumulative values are used on UPDATE (replacing the DB value) while delta values are used on INSERT (starting fresh). Request counts are always deltas, accumulated in DB. All deltas are reset immediately after snapshot so Track() calls durin

(ctx context.Context, db database.Store, replicaID uuid.UUID)

// All deltas are reset immediately after snapshot so Track() calls during the
// DB operation are preserved for the next flush.
func (t *Tracker) FlushToDB(ctx context.Context, db database.Store, replicaID uuid.UUID) error {
	t.mu.Lock()
	if !t.usageSinceLastFlush {
		t.mu.Unlock()
		return nil
	}

	// Snapshot all values.
	workspaceCount := int64(len(t.workspaces))      // cumulative, for UPDATE
	userCount := int64(len(t.users))                // cumulative, for UPDATE
	workspaceDelta := int64(len(t.workspacesDelta)) // delta, for INSERT
	userDelta := int64(len(t.usersDelta))           // delta, for INSERT
	allowed := t.allowedRequests                    // delta, accumulated in DB
	denied := t.deniedRequests                      // delta, accumulated in DB

	// Reset all deltas immediately so Track() calls during the DB operation
	// below are preserved for the next flush.
	t.workspacesDelta = make(map[uuid.UUID]struct{})
	t.usersDelta = make(map[uuid.UUID]struct{})
	t.allowedRequests = 0
	t.deniedRequests = 0
	t.usageSinceLastFlush = false
	t.mu.Unlock()

	//nolint:gocritic // This is the actual package doing boundary usage tracking.
	authCtx := dbauthz.AsBoundaryUsageTracker(ctx)
	err := db.InTx(func(tx database.Store) error {
		// The advisory lock ensures a clean period cutover by preventing
		// this upsert from racing with the aggregate+delete in
		// GetAndResetBoundaryUsageSummary. Without it, upserted data
		// could be lost or miscounted across periods.
		if err := tx.AcquireLock(authCtx, database.LockIDBoundaryUsageStats); err != nil {
			return err
		}
		_, err := tx.UpsertBoundaryUsageStats(authCtx, database.UpsertBoundaryUsageStatsParams{
			ReplicaID:             replicaID,
			UniqueWorkspacesCount: workspaceCount, // cumulative, for UPDATE
			UniqueUsersCount:      userCount,      // cumulative, for UPDATE
			UniqueWorkspacesDelta: workspaceDelta, // delta, for INSERT
			UniqueUsersDelta:      userDelta,      // delta, for INSERT
			AllowedRequests:       allowed,
			DeniedRequests:        denied,
		})
		return err
	}, nil)

	// Always reset cumulative counts to prevent unbounded memory growth (e.g.
	// if the DB is unreachable). Copy delta maps to preserve any Track() calls
	// that occurred during the DB operation above.
	t.mu.Lock()
	t.workspaces = make(map[uuid.UUID]struct{})
	t.users = make(map[uuid.UUID]struct{})
	for id := range t.workspacesDelta {
		t.workspaces[id] = struct{}{}
	}
	for id := range t.usersDelta {
		t.users[id] = struct{}{}
	}