Function TestIntegration_UsagePushAck

internal/nodeagent/integration_test.go:161–231

TestIntegration_UsagePushAck 端到端：node push usage_push → hub 自动 ack → node 调 DoUsage(reset=true) 推进 baseline。

(t *testing.T)

// TestIntegration_UsagePushAck 端到端：node push usage_push → hub 自动 ack
// → node 调 DoUsage(reset=true) 推进 baseline。
func TestIntegration_UsagePushAck(t *testing.T) {
	t.Parallel()
	ph := &integrationPushHandler{}
	_, dialer, stop := startHub(t, ph)
	defer stop()

	usageAPI := &countingUsageAPI{}
	pusher := NewUsagePusher(usageAPI, time.Hour)
	pusher.SetAckTimeout(2 * time.Second)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	cfg := Config{
		NodeID:        "n1",
		ServerAddr:    "passthrough:///bufnet",
		Dispatcher:    NoopDispatcher{},
		HelloProvider: DefaultHelloProvider("n1", nil),
		Dialer: func(_ context.Context) (*grpc.ClientConn, error) {
			return grpc.NewClient("passthrough:///bufnet",
				grpc.WithContextDialer(dialer),
				grpc.WithTransportCredentials(insecure.NewCredentials()),
			)
		},
		OnConnected: func(_ context.Context, s Sender) {
			pusher.SetSender(s)
		},
	}
	done := make(chan error, 1)
	go func() { done <- Run(ctx, cfg) }()
	defer func() { cancel(); <-done }()

	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if pusher.currentSender() != nil {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	if pusher.currentSender() == nil {
		t.Fatalf("sender not injected")
	}

	// 第一次 tick：priming reset，不 push。
	pusher.tick(ctx)
	if got := usageAPI.resets.Load(); got != 1 {
		t.Fatalf("priming reset count=%d want 1", got)
	}

	// 第二次 tick：push；hub 自动 ack；节点端 ack 后再 reset。
	pusher.tick(ctx)

	deadline = time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		_, ups := ph.snapshot()
		if len(ups) >= 1 && pusher.PendingCount() == 0 && usageAPI.resets.Load() >= 2 {
			break
		}