attestor forward same timestamp

services/attestor/config.yaml.example

@@ -22,6 +22,7 @@ attestor:
     - ETH/USD
   polling_time: 300ms
   batch_mode: true
+  use_oracle_timestamp: false  # When true, embed the oracle contract's timestamp in the intent instead of time.Now()
 
 # Logging Configuration
 logging:
@@ -46,6 +47,7 @@ api:
 # - ATTESTOR_ATTESTOR_SYMBOLS (comma-separated)
 # - ATTESTOR_ATTESTOR_POLLING_TIME (e.g., "5s", "300ms")
 # - ATTESTOR_ATTESTOR_BATCH_MODE (true/false)
+# - ATTESTOR_ATTESTOR_USE_ORACLE_TIMESTAMP (true/false) — use oracle contract timestamp instead of time.Now()
 # - ATTESTOR_ATTESTOR_MODE (prime/replica)
 # - ATTESTOR_LOGGING_LEVEL
 # - ATTESTOR_METRICS_PORT

services/attestor/pkg/config/config.go

@@ -110,6 +110,7 @@ type Config struct {
 		DeviationTrigger    bool           `mapstructure:"deviation_trigger"`
 		DeviationThreshold  int            `mapstructure:"deviation_threshold"`
 		ForceUpdateInterval time.Duration  `mapstructure:"force_update_interval"`
+		UseOracleTimestamp  bool           `mapstructure:"use_oracle_timestamp"`
 		Guardian            GuardianConfig `mapstructure:"guardian"`
 	} `mapstructure:"attestor"`
 
@@ -187,6 +188,7 @@ func Init(configPath string) (*Config, error) {
 	v.BindEnv("attestor.deviation_trigger", "ATTESTOR_ATTESTOR_DEVIATION_TRIGGER")
 	v.BindEnv("attestor.deviation_threshold", "ATTESTOR_ATTESTOR_DEVIATION_THRESHOLD")
 	v.BindEnv("attestor.force_update_interval", "ATTESTOR_ATTESTOR_FORCE_UPDATE_INTERVAL")
+	v.BindEnv("attestor.use_oracle_timestamp", "ATTESTOR_ATTESTOR_USE_ORACLE_TIMESTAMP")
 	v.BindEnv("oracle.client_type", "ATTESTOR_ORACLE_CLIENT_TYPE")
 	v.BindEnv("attestor.guardian.default.max_deviation_bips", "ATTESTOR_GUARDIAN_MAX_DEVIATION_BIPS")
 	v.BindEnv("attestor.guardian.default.max_timestamp_age", "ATTESTOR_GUARDIAN_MAX_TIMESTAMP_AGE")
@@ -208,6 +210,7 @@ func Init(configPath string) (*Config, error) {
 	v.SetDefault("attestor.deviation_trigger", false)
 	v.SetDefault("attestor.deviation_threshold", 50)
 	v.SetDefault("attestor.force_update_interval", "0s")
+	v.SetDefault("attestor.use_oracle_timestamp", false)
 	v.SetDefault("attestor.guardian.default.max_deviation_bips", 500)
 	v.SetDefault("attestor.guardian.default.max_timestamp_age", 3600)
 	v.SetDefault("attestor.guardian.default.min_guardian_matches", 1)

services/attestor/pkg/intent/intent.go

@@ -42,20 +42,25 @@ func generateNonce() *big.Int {
 	return big.NewInt(nonce)
 }
 
-func AttestValue(ctx context.Context, multiClient *multirpc.MultiClient, privateKey string, fromAddress string, price *big.Int, volume *big.Int, symbol string) (string, error) {
+func AttestValue(ctx context.Context, multiClient *multirpc.MultiClient, privateKey string, fromAddress string, price *big.Int, volume *big.Int, symbol string, oracleTimestamp *big.Int) (string, error) {
 	if privateKey == "" {
 		return "", fmt.Errorf("private key not provided")
 	}
 	if multiClient == nil {
 		return "", fmt.Errorf("multiClient is required")
 	}
 
-	now := time.Now().Unix()
-	nowBig := big.NewInt(now)
+	// Use oracle timestamp if provided, otherwise fall back to current time
+	var nowBig *big.Int
+	if oracleTimestamp != nil && oracleTimestamp.Sign() > 0 {
+		nowBig = new(big.Int).Set(oracleTimestamp)
+	} else {
+		nowBig = big.NewInt(time.Now().Unix())
+	}
 
 	// Generate unique nonce using atomic counter to prevent collisions
 	nonce := generateNonce()
-	expiry := big.NewInt(now + 3600)
+	expiry := big.NewInt(time.Now().Unix() + 3600)
 
 	cfg := config.Get()
 	intentType := cfg.Attestor.IntentType
@@ -218,9 +223,10 @@ func AttestValue(ctx context.Context, multiClient *multirpc.MultiClient, private
 
 // SymbolData represents price data for a single symbol
 type SymbolData struct {
-	Symbol string
-	Price  *big.Int
-	Volume *big.Int
+	Symbol    string
+	Price     *big.Int
+	Volume    *big.Int
+	Timestamp *big.Int // Oracle timestamp; nil means use time.Now() at signing
 }
 
 func AttestMultipleValues(ctx context.Context, multiClient *multirpc.MultiClient, privateKey string, fromAddress string, symbolsData []SymbolData) (string, error) {
@@ -282,6 +288,12 @@ func AttestMultipleValues(ctx context.Context, multiClient *multirpc.MultiClient
 		// Generate unique nonce for each intent in the batch to prevent collisions
 		nonce := generateNonce()
 
+		// Use per-symbol oracle timestamp if provided, otherwise fall back to current time
+		ts := nowBig
+		if data.Timestamp != nil && data.Timestamp.Sign() > 0 {
+			ts = data.Timestamp
+		}
+
 		intent := types.OracleIntent{
 			IntentType: intentType,
 			Version:    intentVersion,
@@ -290,7 +302,7 @@ func AttestMultipleValues(ctx context.Context, multiClient *multirpc.MultiClient
 			Expiry:     expiry,
 			Symbol:     data.Symbol,
 			Price:      data.Price,
-			Timestamp:  nowBig,
+			Timestamp:  ts,
 			Source:     "DIA Oracle",
 		}
 

services/attestor/pkg/interfaces/signer.go

@@ -8,15 +8,17 @@ import (
 // IntentSigner defines the interface for signing intents
 type IntentSigner interface {
 	// SignIntent creates an EIP-712 signed intent for a single value
-	SignIntent(ctx context.Context, price, volume *big.Int, symbol string) ([]byte, error)
+	// timestamp is the oracle timestamp to embed in the intent; pass nil to use time.Now()
+	SignIntent(ctx context.Context, price, volume, timestamp *big.Int, symbol string) ([]byte, error)
 
 	// SignBatchIntent creates an EIP-712 signed intent for multiple values
 	SignBatchIntent(ctx context.Context, values []SymbolData) ([]byte, error)
 }
 
 // SymbolData represents data for a single symbol in a batch
 type SymbolData struct {
-	Symbol string
-	Price  *big.Int
-	Volume *big.Int
+	Symbol    string
+	Price     *big.Int
+	Volume    *big.Int
+	Timestamp *big.Int // Oracle timestamp; nil means use time.Now() at signing
 }

services/attestor/pkg/service/attestor.go

@@ -348,8 +348,14 @@ func (s *AttestorService) processSingleAttestation(ctx context.Context, symbol s
 		"MinGuardianMatches": guardianParams.MinGuardianMatches,
 	}).Debug("Retrieved oracle value")
 
+	// Determine timestamp to embed in the intent
+	var intentTimestamp *big.Int
+	if s.config.Attestor.UseOracleTimestamp && timestamp != nil && timestamp.Sign() > 0 {
+		intentTimestamp = timestamp
+	}
+
 	// Sign intent
-	signedIntent, err := s.signer.SignIntent(ctx, price, volume, symbol)
+	signedIntent, err := s.signer.SignIntent(ctx, price, volume, intentTimestamp, symbol)
 	if err != nil {
 		s.metrics.RecordIntentCreated(symbol, false)
 		return errors.NewSignerError("sign intent", symbol, err)
@@ -447,10 +453,17 @@ symbolLoop:
 		}
 		logger.WithFields(logFields).Debug("Retrieved oracle value")
 
+		// Set oracle timestamp if configured
+		var ts *big.Int
+		if s.config.Attestor.UseOracleTimestamp && timestamp != nil && timestamp.Sign() > 0 {
+			ts = timestamp
+		}
+
 		symbolData = append(symbolData, interfaces.SymbolData{
-			Symbol: symbol,
-			Price:  price,
-			Volume: volume,
+			Symbol:    symbol,
+			Price:     price,
+			Volume:    volume,
+			Timestamp: ts,
 		})
 	}
 

services/attestor/pkg/signer/eip712.go

@@ -51,7 +51,7 @@ func NewEIP712Signer(privateKeyHex string, rpcURLs []string) (*EIP712Signer, err
 }
 
 // SignIntent creates an EIP-712 signed intent for a single value
-func (s *EIP712Signer) SignIntent(ctx context.Context, price, volume *big.Int, symbol string) ([]byte, error) {
+func (s *EIP712Signer) SignIntent(ctx context.Context, price, volume, timestamp *big.Int, symbol string) ([]byte, error) {
 	// Validate inputs
 	if price == nil || price.Sign() <= 0 {
 		return nil, errors.NewValidationError("price", price, "must be positive")
@@ -63,7 +63,7 @@ func (s *EIP712Signer) SignIntent(ctx context.Context, price, volume *big.Int, s
 		return nil, errors.NewValidationError("symbol", symbol, "must not be empty")
 	}
 
-	signedIntentJSON, err := intent.AttestValue(ctx, s.signingClient, s.privateKeyHex, s.address.Hex(), price, volume, symbol)
+	signedIntentJSON, err := intent.AttestValue(ctx, s.signingClient, s.privateKeyHex, s.address.Hex(), price, volume, symbol, timestamp)
 	if err != nil {
 		return nil, errors.NewSignerError("sign intent", symbol, err)
 	}
@@ -104,9 +104,10 @@ func (s *EIP712Signer) SignBatchIntent(ctx context.Context, values []interfaces.
 		}
 
 		symbolData = append(symbolData, intent.SymbolData{
-			Symbol: v.Symbol,
-			Price:  v.Price,
-			Volume: v.Volume,
+			Symbol:    v.Symbol,
+			Price:     v.Price,
+			Volume:    v.Volume,
+			Timestamp: v.Timestamp,
 		})
 	}
 

services/bridge/internal/api/server.go

@@ -45,6 +45,11 @@ type PoolLister interface {
 	ListPools() []PoolInfo
 }
 
+// QueueStatter provides read-only access to transaction queue stats
+type QueueStatter interface {
+	GetAllQueueStats() []map[string]interface{}
+}
+
 // Server represents the API server
 type Server struct {
 	config         *config.APIConfig
@@ -58,6 +63,7 @@ type Server struct {
 	httpServer      *http.Server
 	failoverHandler *FailoverHandler
 	poolLister      PoolLister
+	queueStatter    QueueStatter
 }
 
 // NewServer creates a new API server
@@ -147,6 +153,11 @@ func (s *Server) SetPoolLister(lister PoolLister) {
 	s.poolLister = lister
 }
 
+// SetQueueStatter sets the queue stats provider for transaction queue inspection
+func (s *Server) SetQueueStatter(statter QueueStatter) {
+	s.queueStatter = statter
+}
+
 // setupRoutes configures all API routes
 func (s *Server) setupRoutes() {
 	s.router.HandleFunc("/health", s.handleHealth).Methods("GET")
@@ -196,6 +207,7 @@ func (s *Server) setupRoutes() {
 	// Worker pool endpoints
 	v1.HandleFunc("/pools", s.handleGetPools).Methods("GET")
 	v1.HandleFunc("/pools/{router_id}/tasks", s.handleGetPoolTasks).Methods("GET")
+	v1.HandleFunc("/queue", s.handleGetQueue).Methods("GET")
 
 	// Failover endpoints (if available)
 	if s.failoverHandler != nil {
@@ -482,6 +494,19 @@ func (s *Server) handleGetPoolTasks(w http.ResponseWriter, r *http.Request) {
 	s.writeError(w, http.StatusNotFound, "Pool not found for router: "+routerID, nil)
 }
 
+func (s *Server) handleGetQueue(w http.ResponseWriter, r *http.Request) {
+	if s.queueStatter == nil {
+		s.writeError(w, http.StatusNotFound, "Queue stats not available", nil)
+		return
+	}
+
+	stats := s.queueStatter.GetAllQueueStats()
+	s.writeJSON(w, map[string]interface{}{
+		"queues": stats,
+		"count":  len(stats),
+	})
+}
+
 // Middleware
 
 func (s *Server) loggingMiddleware(next http.Handler) http.Handler {