Deadline Propagation

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

// Deadline propagation implementation
import { Request, Response, NextFunction } from 'express';
 
const DEADLINE_HEADER = 'x-request-deadline';
const DEFAULT_DEADLINE_MS = 30000;
 
interface DeadlineContext {
  deadline: Date;
  isExpired(): boolean;
  getRemainingMs(): number;
  toISOString(): string;
}
 
// Create deadline context from absolute deadline
function createDeadlineContext(deadline: Date): DeadlineContext {
  return {
    deadline,
    isExpired: () => Date.now() >= deadline.getTime(),
    getRemainingMs: () => Math.max(0, deadline.getTime() - Date.now()),
    toISOString: () => deadline.toISOString(),
  };
}
 
// Middleware: Extract or establish deadline
export function deadlineMiddleware(
  req: Request,
  res: Response,
  next: NextFunction
): void {
  let deadline: Date;
  
  // Check for incoming deadline header
  const deadlineHeader = req.get(DEADLINE_HEADER);
  if (deadlineHeader) {
    const parsed = Date.parse(deadlineHeader);
    if (!isNaN(parsed)) {
      deadline = new Date(parsed);
    } else {
      // Invalid deadline, establish new one
      deadline = new Date(Date.now() + DEFAULT_DEADLINE_MS);
    }
  } else {
    // No deadline provided, establish one
    deadline = new Date(Date.now() + DEFAULT_DEADLINE_MS);
  }
  
  // Check if deadline is already expired
  if (Date.now() >= deadline.getTime()) {
    res.status(504).json({ 
      error: 'Deadline exceeded before processing',
      deadline: deadline.toISOString(),
    });
    return;
  }
  
  // Attach context to request
  req.deadlineContext = createDeadlineContext(deadline);
  
  // Set response timeout
  const remainingMs = req.deadlineContext.getRemainingMs();
  res.setTimeout(remainingMs);
  
  next();
}
 
// Decorator: Deadline-aware function execution
async function withDeadline<T>(
  deadline: DeadlineContext,
  operation: () => Promise<T>,
  operationName: string = 'operation'
): Promise<T> {
  // Check deadline before starting
  if (deadline.isExpired()) {
    throw new DeadlineExceededError(
      `Deadline expired before starting ${operationName}`,
      deadline.deadline
    );
  }
  
  const remainingMs = deadline.getRemainingMs();
  
  // Create timeout race
  return Promise.race([
    operation(),
    new Promise<never>((_, reject) => {
      setTimeout(() => {
        reject(new DeadlineExceededError(
          `Deadline exceeded during ${operationName}`,
          deadline.deadline
        ));
      }, remainingMs);
    }),
  ]);
}
 
// HTTP client that propagates deadlines
class DeadlineAwareHttpClient {
  constructor(private baseUrl: string) {}
  
  async request(
    path: string,
    options: RequestInit,
    deadline: DeadlineContext
  ): Promise<Response> {
    // Check if we have enough time
    const remainingMs = deadline.getRemainingMs();
    const minRequired = 100; // Minimum ms to attempt a request
    
    if (remainingMs < minRequired) {
      throw new DeadlineExceededError(
        'Insufficient time for downstream request',
        deadline.deadline
      );
    }
    
    // Add deadline header
    const headers = new Headers(options.headers);
    headers.set(DEADLINE_HEADER, deadline.toISOString());
    
    // Create abort controller for timeout
    const controller = new AbortController();
    const timeoutId = setTimeout(() => controller.abort(), remainingMs);
    
    try {
      return await fetch(`${this.baseUrl}${path}`, {
        ...options,
        headers,
        signal: controller.signal,
      });
    } finally {
      clearTimeout(timeoutId);
    }
  }
}
 
// Custom error class
class DeadlineExceededError extends Error {
  constructor(
    message: string,
    public readonly deadline: Date,
    public readonly exceededBy: number = Date.now() - deadline.getTime()
  ) {
    super(message);
    this.name = 'DeadlineExceededError';
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

// Skew-tolerant deadline handling
const EXPECTED_CLOCK_SKEW_MS = 50; // Conservative estimate for NTP-synced servers
 
interface SkewTolerantDeadlineContext {
  deadline: Date;
  assumedSkewMs: number;
  
  // Conservative remaining time (accounts for possible skew)
  getRemainingMsConservative(): number;
  
  // Optimistic remaining time (ignores skew)
  getRemainingMsOptimistic(): number;
  
  // Check if deadline is definitely expired (even with skew)
  isDefinitelyExpired(): boolean;
  
  // Check if deadline might be expired (considering skew)
  isPossiblyExpired(): boolean;
}
 
function createSkewTolerantContext(
  deadline: Date,
  assumedSkewMs: number = EXPECTED_CLOCK_SKEW_MS
): SkewTolerantDeadlineContext {
  return {
    deadline,
    assumedSkewMs,
    
    getRemainingMsConservative() {
      // Assume our clock might be slow, so deadline might be sooner
      return Math.max(0, deadline.getTime() - Date.now() - assumedSkewMs);
    },
    
    getRemainingMsOptimistic() {
      // Assume our clock might be fast, so we might have more time
      return Math.max(0, deadline.getTime() - Date.now() + assumedSkewMs);
    },
    
    isDefinitelyExpired() {
      // Even if our clock is maximally fast, deadline has passed
      return Date.now() - assumedSkewMs >= deadline.getTime();
    },
    
    isPossiblyExpired() {
      // Our clock might be slow, so deadline might have passed
      return Date.now() + assumedSkewMs >= deadline.getTime();
    },
  };
}
 
// Usage example
async function handleWithSkewTolerance(
  ctx: SkewTolerantDeadlineContext,
  operation: () => Promise<void>
): Promise<void> {
  // Use conservative estimate for timeouts
  const conservativeRemaining = ctx.getRemainingMsConservative();
  
  if (conservativeRemaining < 100) {
    throw new Error('Deadline too close (with clock skew consideration)');
  }
  
  // Log the uncertainty
  console.log(`Deadline remaining: ${ctx.getRemainingMsOptimistic()}ms to ${conservativeRemaining}ms (±${ctx.assumedSkewMs}ms)`);
  
  // Set timeout using conservative estimate
  const timeoutHandle = setTimeout(() => {
    throw new DeadlineExceededError('Conservative deadline exceeded', ctx.deadline);
  }, conservativeRemaining);
  
  try {
    await operation();
  } finally {
    clearTimeout(timeoutHandle);
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

package main
 
import (
    "context"
    "log"
    "time"
    
    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
    pb "example.com/order/proto"
)
 
func main() {
    conn, err := grpc.Dial("order-service:50051", grpc.WithInsecure())
    if err != nil {
        log.Fatalf("Failed to connect: %v", err)
    }
    defer conn.Close()
    
    client := pb.NewOrderServiceClient(conn)
    
    // Option 1: Set deadline (absolute)
    deadline := time.Now().Add(5 * time.Second)
    ctx, cancel := context.WithDeadline(context.Background(), deadline)
    defer cancel()
    
    // Option 2: Set timeout (relative, converted to deadline internally)
    // ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    
    // Make the RPC - deadline is automatically included
    resp, err := client.PlaceOrder(ctx, &pb.OrderRequest{
        Items: []*pb.Item{{Id: "item-1", Quantity: 2}},
    })
    
    if err != nil {
        // Check for deadline exceeded
        if status.Code(err) == codes.DeadlineExceeded {
            log.Println("Order placement timed out")
            return
        }
        // Check if our local context was cancelled
        if ctx.Err() == context.DeadlineExceeded {
            log.Println("Client deadline exceeded before response")
            return
        }
        log.Fatalf("Order failed: %v", err)
    }
    
    log.Printf("Order placed: %v", resp.OrderId)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

interface RecommendationOptions {
  fast: { maxLatencyMs: 100, quality: 'basic' };
  standard: { maxLatencyMs: 500, quality: 'good' };
  comprehensive: { maxLatencyMs: 2000, quality: 'best' };
}
 
async function getRecommendations(
  userId: string,
  deadline: DeadlineContext
): Promise<Recommendations> {
  const remainingMs = deadline.getRemainingMs();
  
  // Select strategy based on available time
  if (remainingMs >= 2000) {
    // Plenty of time - use ML model with full feature set
    return comprehensiveRecommendations(userId);
  } else if (remainingMs >= 500) {
    // Moderate time - use cached model results + light personalization
    return standardRecommendations(userId);
  } else if (remainingMs >= 100) {
    // Limited time - return cached popular items
    return fastRecommendations(userId);
  } else {
    // No time - skip recommendations entirely
    return { items: [], source: 'skipped-due-to-deadline' };
  }
}
 
// In the handler:
async function productPageHandler(req: Request): Promise<Response> {
  const product = await getProduct(req.productId); // Always needed
  
  // Only include recommendations if we have time
  const remainingAfterProduct = req.deadlineContext.getRemainingMs();
  const recommendations = await getRecommendations(
    req.userId,
    req.deadlineContext
  );
  
  return {
    product,
    recommendations,
  };
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

interface RetryConfig {
  maxAttempts: number;
  initialDelayMs: number;
  maxDelayMs: number;
  backoffMultiplier: number;
}
 
async function executeWithDeadlineAwareRetry<T>(
  operation: () => Promise<T>,
  config: RetryConfig,
  deadline: DeadlineContext,
  minTimeForAttemptMs: number = 500
): Promise<T> {
  let lastError: Error | null = null;
  let delay = config.initialDelayMs;
  
  for (let attempt = 1; attempt <= config.maxAttempts; attempt++) {
    const remainingMs = deadline.getRemainingMs();
    
    // Check if we have enough time for another attempt
    if (remainingMs < minTimeForAttemptMs) {
      throw new DeadlineExceededError(
        `Not enough time for attempt ${attempt}. Remaining: ${remainingMs}ms, Required: ${minTimeForAttemptMs}ms`,
        deadline.deadline
      );
    }
    
    try {
      // Cap operation timeout at remaining time
      const attemptTimeout = Math.min(
        remainingMs - 100, // Leave buffer for retry logic
        5000 // Maximum per-attempt timeout
      );
      
      return await Promise.race([
        operation(),
        new Promise<never>((_, reject) => 
          setTimeout(() => reject(new Error('Attempt timeout')), attemptTimeout)
        ),
      ]);
    } catch (error) {
      lastError = error as Error;
      
      // Check if we should retry
      const remainingAfterAttempt = deadline.getRemainingMs();
      const nextDelay = Math.min(delay, remainingAfterAttempt - minTimeForAttemptMs);
      
      if (
        attempt < config.maxAttempts &&
        nextDelay > 0 &&
        remainingAfterAttempt > minTimeForAttemptMs + nextDelay
      ) {
        console.log(
          `Attempt ${attempt} failed, retrying in ${nextDelay}ms. ` +
          `Remaining time: ${remainingAfterAttempt}ms`
        );
        await sleep(nextDelay);
        delay = Math.min(delay * config.backoffMultiplier, config.maxDelayMs);
      } else {
        // No more time for retries
        break;
      }
    }
  }
  
  throw lastError || new Error('All retry attempts exhausted');
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

interface QueueItem<T> {
  data: T;
  deadline: Date;
  enqueuedAt: Date;
}
 
class DeadlineAwarePriorityQueue<T> {
  private items: QueueItem<T>[] = [];
  
  enqueue(data: T, deadline: Date): void {
    this.items.push({
      data,
      deadline,
      enqueuedAt: new Date(),
    });
    // Sort by deadline (earliest first)
    this.items.sort((a, b) => a.deadline.getTime() - b.deadline.getTime());
  }
  
  dequeue(): QueueItem<T> | null {
    // Remove expired items
    const now = Date.now();
    this.items = this.items.filter(item => item.deadline.getTime() > now);
    
    // Return the item with earliest deadline
    return this.items.shift() || null;
  }
  
  // Get item with tightest deadline that has enough time to process
  dequeueWithMinTime(minRequiredMs: number): QueueItem<T> | null {
    const now = Date.now();
    
    for (let i = 0; i < this.items.length; i++) {
      const item = this.items[i];
      const remaining = item.deadline.getTime() - now;
      
      if (remaining >= minRequiredMs) {
        this.items.splice(i, 1);
        return item;
      }
    }
    
    return null;
  }
  
  // Metrics for monitoring
  getStats(): { total: number; expired: number; byUrgency: Record<string, number> } {
    const now = Date.now();
    let expired = 0;
    const byUrgency = { critical: 0, urgent: 0, normal: 0 };
    
    for (const item of this.items) {
      const remaining = item.deadline.getTime() - now;
      if (remaining <= 0) {
        expired++;
      } else if (remaining < 1000) {
        byUrgency.critical++;
      } else if (remaining < 5000) {
        byUrgency.urgent++;
      } else {
        byUrgency.normal++;
      }
    }
    
    return { total: this.items.length, expired, byUrgency };
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

package main
 
import (
    "context"
    "database/sql"
    "errors"
    "log"
    "time"
)
 
func processOrder(ctx context.Context, orderID string) error {
    // Context carries the deadline; all operations check it
    
    // 1. Database query - will be cancelled if deadline expires
    var order Order
    err := db.QueryRowContext(ctx,
        "SELECT * FROM orders WHERE id = ?", orderID,
    ).Scan(&order.ID, &order.Status)
    
    if err != nil {
        if errors.Is(err, context.DeadlineExceeded) {
            log.Printf("Query cancelled due to deadline")
            return err
        }
        return err
    }
    
    // 2. HTTP call - will be cancelled if deadline expires
    req, _ := http.NewRequestWithContext(ctx, "POST", paymentURL, body)
    resp, err := httpClient.Do(req)
    if err != nil {
        if errors.Is(err, context.DeadlineExceeded) {
            log.Printf("Payment call cancelled due to deadline")
            // Note: Payment might have succeeded - need idempotency handling
            return err
        }
        return err
    }
    defer resp.Body.Close()
    
    // 3. Long computation - periodically check context
    result, err := computeExpensiveResult(ctx, order)
    if err != nil {
        return err
    }
    
    return nil
}
 
// Long operation that respects context cancellation
func computeExpensiveResult(ctx context.Context, order Order) (*Result, error) {
    result := &Result{}
    
    for i := 0; i < 1000; i++ {
        // Check for cancellation periodically
        select {
        case <-ctx.Done():
            return nil, ctx.Err()
        default:
        }
        
        // Do some work
        partial := processOrderItem(order.Items[i%len(order.Items)])
        result.Merge(partial)
    }
    
    return result, nil
}
 
// Transaction with deadline-aware rollback
func processOrderWithTransaction(ctx context.Context, orderID string) error {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    
    // Defer rollback - will execute if we don't commit
    defer func() {
        if tx != nil {
            tx.Rollback()
        }
    }()
    
    // Execute operations within transaction
    if _, err := tx.ExecContext(ctx, 
        "UPDATE inventory SET reserved = reserved + ? WHERE item_id = ?",
        order.Quantity, order.ItemID,
    ); err != nil {
        return err // Rollback happens via defer
    }
    
    // Check if deadline is approaching before committing
    if deadline, ok := ctx.Deadline(); ok {
        if time.Until(deadline) < 100*time.Millisecond {
            // Not enough time to safely commit
            return context.DeadlineExceeded
        }
    }
    
    // Commit the transaction
    if err := tx.Commit(); err != nil {
        return err
    }
    
    tx = nil // Prevent rollback in defer
    return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

async function processWithDeadline(
  deadline: Date,
  operation: (signal: AbortSignal) => Promise<void>
): Promise<void> {
  // Create abort controller
  const controller = new AbortController();
  const { signal } = controller;
  
  // Calculate remaining time
  const remainingMs = Math.max(0, deadline.getTime() - Date.now());
  
  // Set up deadline timeout
  const timeoutId = setTimeout(() => {
    controller.abort(new DeadlineExceededError('Deadline exceeded', deadline));
  }, remainingMs);
  
  try {
    await operation(signal);
  } finally {
    clearTimeout(timeoutId);
  }
}
 
// Usage:
await processWithDeadline(deadline, async (signal) => {
  // Fetch with abort signal
  const response = await fetch('/api/data', { signal });
  
  // The fetch will be aborted if deadline is reached
  if (signal.aborted) {
    throw signal.reason;
  }
  
  const data = await response.json();
  
  // Check periodically in loops
  for (const item of items) {
    if (signal.aborted) {
      throw signal.reason;
    }
    await processItem(item, signal);
  }
});