fix: rebuild the ADC baseline on leader acquisition

Makefile

@@ -30,7 +30,7 @@ KIND_NAME ?= apisix-ingress-cluster
 KIND_NODE_IMAGE ?= kindest/node:v1.30.0@sha256:047357ac0cfea04663786a612ba1eaba9702bef25227a794b52890dd8bcd692e
 
 DASHBOARD_VERSION ?= dev
-ADC_VERSION ?= 0.22.1
+ADC_VERSION ?= 0.27.1
 
 DIR := $(shell pwd)
 

api/adc/types.go

@@ -789,6 +789,11 @@ type Config struct {
 	Token       string
 	TlsVerify   bool
 	BackendType string
+
+	// BypassCache makes the ADC server drop the in-memory baseline it holds for this
+	// cacheKey and re-derive it from the data plane before computing the diff. It is a
+	// per-request flag set on the sync path, not part of the translated configuration.
+	BypassCache bool
 }
 
 // MarshalJSON implements custom JSON marshaling for adcConfig

docs/en/latest/developer-guide.md

@@ -36,7 +36,7 @@ Before you get started make sure you have:
 1. Installed [Go 1.23](https://golang.org/dl/) or later
 2. A Kubernetes cluster available. We recommend using [kind](https://kind.sigs.k8s.io/).
 3. Installed APISIX in Kubernetes using [Helm](https://github.com/apache/apisix-helm-chart).
-4. Installed [ADC v0.20.0+](https://github.com/api7/adc/releases)
+4. Installed [ADC v0.27.0+](https://github.com/api7/adc/releases)
 
 ## Fork and clone
 

internal/adc/client/client.go

@@ -50,6 +50,13 @@ type Client struct {
 
 	defaultMode string
 
+	// rebuiltMu guards rebuiltBaselines.
+	rebuiltMu sync.Mutex
+	// rebuiltBaselines holds the cacheKeys whose ADC baseline this leadership term has
+	// already re-derived from the data plane. A key missing from it is synced with
+	// bypassCache first. See InvalidateADCCache.
+	rebuiltBaselines map[string]struct{}
+
 	log logr.Logger
 }
 
@@ -67,6 +74,7 @@ func New(log logr.Logger, defaultMode string, timeout time.Duration) (*Client, e
 
 	return &Client{
 		Store:            store,
+		rebuiltBaselines: make(map[string]struct{}),
 		executor:         NewHTTPADCExecutor(log, serverURL, timeout),
 		ConfigManager:    configManager,
 		ADCDebugProvider: common.NewADCDebugProvider(store, configManager),
@@ -75,6 +83,54 @@ func New(log logr.Logger, defaultMode string, timeout time.Duration) (*Client, e
 	}, nil
 }
 
+// InvalidateADCCache forgets which ADC baselines are known to be current, so that the
+// next sync of each cacheKey re-derives its baseline from the data plane.
+//
+// It is called on leader acquisition, which is the one moment a stale baseline can enter
+// the picture. The ADC server is a sidecar that outlives the controller process: losing
+// the lease terminates the manager container but not the sidecar, so what ADC holds for a
+// cacheKey -- the last synced content plus the conf_version it generated -- can still be
+// the snapshot this pod left behind in an earlier term, while the leader in between kept
+// pushing and moved the data plane's conf_version past it. APISIX standalone requires
+// those versions to be monotonic and refuses the whole configuration otherwise.
+func (c *Client) InvalidateADCCache() {
+	c.rebuiltMu.Lock()
+	defer c.rebuiltMu.Unlock()
+	clear(c.rebuiltBaselines)
+}
+
+func (c *Client) baselineIsCurrent(cacheKey string) bool {
+	c.rebuiltMu.Lock()
+	defer c.rebuiltMu.Unlock()
+	_, ok := c.rebuiltBaselines[cacheKey]
+	return ok
+}
+
+func (c *Client) markBaselineCurrent(cacheKey string) {
+	c.rebuiltMu.Lock()
+	defer c.rebuiltMu.Unlock()
+	c.rebuiltBaselines[cacheKey] = struct{}{}
+}
+
+// isConfVersionRejection reports whether the data plane refused the push because of a
+// conf_version, which is the one rejection re-deriving the baseline can answer.
+//
+// It matches the field name, not the sentence. conf_version is part of the standalone
+// admin API -- we send those keys ourselves -- so any rejection that concerns it names it,
+// whatever prose APISIX wraps it in. Matching the sentence would tie us to prose APISIX is
+// free to reword; matching the field only breaks if it renames the API.
+//
+// This backs the safety net, not the fix. A baseline is rebuilt on leader acquisition,
+// which is where staleness comes from, so if this ever stopped firing the reported bug
+// would not come back with it.
+func isConfVersionRejection(err error) bool {
+	return err != nil && strings.Contains(err.Error(), confVersionField)
+}
+
+// confVersionField names the monotonic version APISIX standalone keeps per resource type
+// (routes_conf_version, upstreams_conf_version, ...) and refuses a push that moves back.
+const confVersionField = "conf_version"
+
 type Task struct {
 	Key           types.NamespacedNameKind
 	Name          string
@@ -264,6 +320,56 @@ func (c *Client) Sync(ctx context.Context) (map[string]types.ADCExecutionErrors,
 	return failedMap, err
 }
 
+// push syncs one config through the ADC server, re-deriving the baseline ADC diffs against
+// whenever that baseline cannot be trusted. Beside the error to report it returns the ones
+// to report next to it, which a rebuild that failed leaves behind.
+//
+// The ADC sidecar outlives the controller process, so the baseline it holds for a cacheKey
+// may be one an earlier leadership term left behind. It is re-derived from the data plane
+// the first time this term syncs the key, before anything can be pushed from it, and only
+// a sync ADC accepts settles the question.
+//
+// Rebuilding on leader acquisition covers where staleness comes from. The safety net covers
+// what it cannot foresee -- another writer on this data plane, a desync no leadership change
+// explains -- and a conf_version the data plane refuses is the only way any of that shows
+// itself. Re-read the data plane and push again.
+func (c *Client) push(ctx context.Context, config adctypes.Config, args []string) ([]types.ADCExecutionError, error) {
+	standalone := config.BackendType == backendAPISIXStandalone
+	config.BypassCache = standalone && !c.baselineIsCurrent(config.Name)
+
+	err := c.executor.Execute(ctx, config, args)
+
+	var alsoReport []types.ADCExecutionError
+	if standalone && !config.BypassCache && isConfVersionRejection(err) {
+		c.log.Info("data plane rejected a stale conf_version, rebuilding the ADC baseline",
+			"config", config.Name, "error", err.Error())
+		// Keep the rejection visible even when the sync recovers. The rebuild is not rate
+		// limited, so a rejection on every sync -- someone else writing to this data plane --
+		// turns every sync into a full fetch and diff, and this counter is what says so.
+		pkgmetrics.RecordExecutionError(config.Name, "conf_version_conflict")
+
+		config.BypassCache = true
+		retryErr := c.executor.Execute(ctx, config, args)
+
+		// Report the rejection as well. On its own a failed rebuild says nothing about what it
+		// was rebuilding for, and it is the rejection that names the cause -- an ADC server too
+		// old to know bypassCache, say, answers with a schema error that points nowhere near
+		// it. Unless the rebuild was rejected the same way, in which case saying it twice only
+		// pads the status message.
+		var rejected types.ADCExecutionError
+		if retryErr != nil && retryErr.Error() != err.Error() && errors.As(err, &rejected) {
+			alsoReport = append(alsoReport, rejected)
+		}
+		err = retryErr
+	}
+
+	// Only a sync ADC accepted proves its baseline is now derived from the data plane.
+	if err == nil && config.BypassCache {
+		c.markBaselineCurrent(config.Name)
+	}
+	return alsoReport, err
+}
+
 func (c *Client) sync(ctx context.Context, task Task) error {
 	c.log.V(1).Info("syncing resources", "task", task)
 
@@ -376,7 +482,9 @@ func (c *Client) sync(ctx context.Context, task Task) error {
 			config.BackendType = c.defaultMode
 		}
 
-		err := c.executor.Execute(ctx, config, args)
+		alsoReport, err := c.push(ctx, config, args)
+		errs.Errors = append(errs.Errors, alsoReport...)
+
 		duration := time.Since(startTime).Seconds()
 
 		status := adctypes.StatusSuccess

internal/adc/client/executor.go

@@ -39,6 +39,11 @@ import (
 
 const (
 	defaultHTTPADCExecutorAddr = "http://127.0.0.1:3000"
+
+	pathSync     = "/sync"
+	pathValidate = "/validate"
+
+	backendAPISIXStandalone = "apisix-standalone"
 )
 
 type ADCExecutor interface {
@@ -80,6 +85,11 @@ type ADCServerOpts struct {
 	IncludeResourceType []string          `json:"includeResourceType,omitempty"`
 	TlsSkipVerify       *bool             `json:"tlsSkipVerify,omitempty"`
 	CacheKey            string            `json:"cacheKey"`
+	// BypassCache is only accepted by the /sync task of ADC >= 0.27.0. Both ADC task
+	// schemas reject unknown fields, so omitempty is what keeps every other request --
+	// /validate, and every sync that is not recovering from a rejection -- byte for byte
+	// what an older ADC server already accepts.
+	BypassCache bool `json:"bypassCache,omitempty"`
 }
 
 type ADCValidateResult struct {
@@ -141,7 +151,7 @@ func (e *HTTPADCExecutor) runHTTPSync(ctx context.Context, config adctypes.Confi
 	}
 
 	serverAddrs := func() []string {
-		if config.BackendType == "apisix-standalone" {
+		if config.BackendType == backendAPISIXStandalone {
 			return []string{strings.Join(config.ServerAddrs, ",")}
 		}
 		return config.ServerAddrs
@@ -218,7 +228,7 @@ func (e *HTTPADCExecutor) runHTTPSyncForSingleServer(ctx context.Context, server
 	}
 
 	// Build HTTP request
-	req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels, types, resources, http.MethodPut, "/sync")
+	req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels, types, resources, http.MethodPut, pathSync)
 	if err != nil {
 		return fmt.Errorf("failed to build HTTP request: %w", err)
 	}
@@ -252,7 +262,7 @@ func (e *HTTPADCExecutor) runHTTPValidateForSingleServer(ctx context.Context, se
 		return fmt.Errorf("failed to load resources from file %s: %w", filePath, err)
 	}
 
-	req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels, types, resources, http.MethodPut, "/validate")
+	req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels, types, resources, http.MethodPut, pathValidate)
 	if err != nil {
 		return fmt.Errorf("failed to build validate request: %w", err)
 	}
@@ -326,6 +336,7 @@ func (e *HTTPADCExecutor) loadResourcesFromFile(filePath string) (*adctypes.Reso
 func (e *HTTPADCExecutor) buildHTTPRequest(ctx context.Context, serverAddr string, config adctypes.Config, labels map[string]string, types []string, resources *adctypes.Resources, method string, path string) (*http.Request, error) {
 	// Prepare request body
 	tlsVerify := config.TlsVerify
+	bypassCache := path == pathSync && config.BypassCache
 	reqBody := ADCServerRequest{
 		Task: ADCServerTask{
 			Opts: ADCServerOpts{
@@ -336,6 +347,7 @@ func (e *HTTPADCExecutor) buildHTTPRequest(ctx context.Context, serverAddr strin
 				IncludeResourceType: types,
 				TlsSkipVerify:       ptr.To(!tlsVerify),
 				CacheKey:            config.Name,
+				BypassCache:         bypassCache,
 			},
 			Config: *resources,
 		},
@@ -353,6 +365,7 @@ func (e *HTTPADCExecutor) buildHTTPRequest(ctx context.Context, serverAddr strin
 		"server", serverAddr,
 		"mode", config.BackendType,
 		"cacheKey", config.Name,
+		"bypassCache", bypassCache,
 		"labelSelector", labels,
 		"includeResourceType", types,
 		"tlsSkipVerify", !tlsVerify,