Resource cleanup misses fast deletions between informer LIST and WATCH and times out with no remaining objects #319

param(
    [Parameter(Mandatory)][string]$DcpPath,
    [string]$PythonPath = 'python',
    [string]$EvidenceDirectory = (Join-Path $PSScriptRoot 'runs')
)
$ErrorActionPreference='Stop'
$runPath=Join-Path $EvidenceDirectory ('native-delay-watch-'+[Guid]::NewGuid().ToString('N'))
New-Item -ItemType Directory $runPath | Out-Null
$env:DCP_DIAGNOSTICS_LOG_FOLDER=$runPath
$env:DCP_DIAGNOSTICS_LOG_LEVEL='debug'
$kubeconfigPath=Join-Path $runPath 'kubeconfig'
$native=Start-Process -FilePath $DcpPath -ArgumentList @('start-apiserver','--monitor', $PID, '--kubeconfig', $kubeconfigPath) -WindowStyle Hidden -PassThru -RedirectStandardOutput (Join-Path $runPath 'stdout.log') -RedirectStandardError (Join-Path $runPath 'stderr.log')
Write-Output "Native PID=$($native.Id) evidence=$runPath"
try {
    $deadline=[DateTime]::UtcNow.AddSeconds(20)
    while (!(Test-Path $kubeconfigPath)) { if ([DateTime]::UtcNow -gt $deadline) { throw 'Native startup did not produce kubeconfig' }; Start-Sleep -Milliseconds 100 }
    $services=@{apiVersion='usvc-dev.developer.microsoft.com/v1';kind='ServiceList';items=@(1..5 | ForEach-Object { @{apiVersion='usvc-dev.developer.microsoft.com/v1';kind='Service';metadata=@{name="probe-$_"};spec=@{protocol='TCP';addressAllocationMode='Localhost'}}})} | ConvertTo-Json -Depth 9
    $services | kubectl --kubeconfig $kubeconfigPath create -f -
    if ($LASTEXITCODE -ne 0) { throw 'Service creation failed' }
    kubectl --kubeconfig $kubeconfigPath wait services.usvc-dev.developer.microsoft.com --all --for=jsonpath='{.metadata.finalizers}' --timeout=20s
    if ($LASTEXITCODE -ne 0) { throw 'Service startup failed' }
    $server=((Get-Content $kubeconfigPath | Select-String '^    server: ').Line -replace '^    server: ','')
    $token=((Get-Content $kubeconfigPath | Select-String '^    token: ').Line -replace '^    token: ','')
    $originalKubeconfigPath=Join-Path $runPath 'original-kubeconfig'
    Copy-Item $kubeconfigPath $originalKubeconfigPath
    $proxyPortPath=Join-Path $runPath 'proxy-port.txt'
    $proxyLogPath=Join-Path $runPath 'proxy.log'
    $proxyScript=Join-Path $runPath 'delay-watch-proxy.py'
@'
import http.client
import http.server
import ssl
import sys
import threading
import time
import urllib.parse

upstream = urllib.parse.urlparse(sys.argv[1])
log_path = sys.argv[3]
log_lock = threading.Lock()
with open(sys.argv[4], encoding='utf-8-sig') as stream:
    token = next(line.strip().removeprefix('token: ') for line in stream if line.strip().startswith('token: '))

def log(value):
    with log_lock:
        with open(log_path, 'a', encoding='utf-8') as stream:
            stream.write(f'{time.time():.6f} {value}\n')

class Proxy(http.server.BaseHTTPRequestHandler):
    protocol_version = 'HTTP/1.1'
    def do_GET(self): self.forward()
    def do_DELETE(self): self.forward()
    def do_PATCH(self): self.forward()
    def do_POST(self): self.forward()
    def do_PUT(self): self.forward()
    def log_message(self, *args): pass
    def forward(self):
        params = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
        watch = self.path.startswith('/apis/usvc-dev.developer.microsoft.com/v1/services?') and params.get('watch') == ['true']
        log(f'{self.command} {self.path} delay={watch}')
        if watch:
            time.sleep(2)
            log(f'forward delayed watch {self.path}')
        length = int(self.headers.get('Content-Length', 0))
        body = self.rfile.read(length) if length else None
        headers = {key: value for key, value in self.headers.items() if key.lower() not in ['host', 'connection', 'accept-encoding']}
        headers['Authorization'] = 'Bearer ' + token
        connection = http.client.HTTPSConnection(upstream.hostname, upstream.port, context=ssl._create_unverified_context(), timeout=180)
        try:
            connection.request(self.command, self.path, body, headers)
            response = connection.getresponse()
            log(f'{self.command} {self.path} response={response.status}')
            self.send_response(response.status)
            for key, value in response.getheaders():
                if key.lower() not in ['connection', 'transfer-encoding']:
                    self.send_header(key, value)
            self.send_header('Connection', 'close')
            self.end_headers()
            while chunk := response.read1(65536):
                self.wfile.write(chunk)
                self.wfile.flush()
        except (BrokenPipeError, ConnectionResetError, TimeoutError, ssl.SSLError) as error:
            log(f'{self.command} {self.path} transport={type(error).__name__}')
        finally:
            self.close_connection = True
            connection.close()

server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Proxy)
server.daemon_threads = True
with open(sys.argv[2], 'w', encoding='utf-8') as stream:
    stream.write(str(server.server_port))
server.serve_forever()

'@ | Set-Content $proxyScript
    $proxy=Start-Process -FilePath $PythonPath -ArgumentList @($proxyScript,$server,$proxyPortPath,$proxyLogPath,$originalKubeconfigPath) -WindowStyle Hidden -PassThru -RedirectStandardOutput (Join-Path $runPath 'proxy-stdout.log') -RedirectStandardError (Join-Path $runPath 'proxy-stderr.log')
    while (!(Test-Path $proxyPortPath)) { Start-Sleep -Milliseconds 50 }
    $proxyPort=Get-Content $proxyPortPath -Raw
    (Get-Content $kubeconfigPath -Raw).Replace($server,"http://127.0.0.1:$proxyPort") | Set-Content $kubeconfigPath
    $headers=@{}; if ($token) { $headers.Authorization="Bearer $token" }
    Invoke-RestMethod "$server/admin/execution" -Method Patch -ContentType 'application/merge-patch+json' -Body '{"status":"CleaningResources","shutdownResourceCleanup":"Full"}' -SkipCertificateCheck -Headers $headers | Select-Object status,shutdownResourceCleanup | ConvertTo-Json -Compress
    $deadline=[DateTime]::UtcNow.AddSeconds(140)
    do {
        $state=Invoke-RestMethod "$server/admin/execution" -SkipCertificateCheck -Headers $headers
        $objects=kubectl --kubeconfig $originalKubeconfigPath get services.usvc-dev.developer.microsoft.com -o json --request-timeout=3s | ConvertFrom-Json
        @{time=[DateTimeOffset]::UtcNow.ToString('O');status=$state.status;count=@($objects.items).Count;items=@($objects.items | ForEach-Object { @{name=$_.metadata.name;finalizers=$_.metadata.finalizers;deletionTimestamp=$_.metadata.deletionTimestamp}})} | ConvertTo-Json -Compress -Depth 7 | Add-Content (Join-Path $runPath 'state.ndjson')
        if ($state.status -eq 'CleanupComplete') { break }
        Start-Sleep -Seconds 1
    } while ([DateTime]::UtcNow -lt $deadline)
    Invoke-RestMethod "$server/admin/execution" -Method Patch -ContentType 'application/merge-patch+json' -Body '{"status":"Stopping","shutdownResourceCleanup":"Full"}' -SkipCertificateCheck -Headers $headers | Out-Null
    $native.WaitForExit(10000) | Out-Null
    Write-Output "Native exited=$($native.HasExited) exit=$($native.ExitCode) state=$($state.status) objects=$(@($objects.items).Count)"
} finally {
    if ($proxy -and !$proxy.HasExited) { Stop-Process -Id $proxy.Id }
    if (!$native.HasExited) { Write-Output "Owned process still running PID=$($native.Id); monitor will stop it at script exit" }
}