从能运行到能放一周：给 ESP32 项目做故障注入清单

// ---- main/main.c ----
/* SPDX-License-Identifier: MIT */
#include <stdio.h>
#include <string.h>
#include <stdint.h>
#include <inttypes.h>
#include <fcntl.h>
#include <stdlib.h>
#include <stdatomic.h>
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/queue.h"
#include "esp_event.h"
#include "esp_netif.h"
#include "esp_wifi.h"
#include "esp_timer.h"
#include "esp_log.h"
#include "esp_system.h"
#include "esp_random.h"
#include "esp_mac.h"
#include "esp_heap_caps.h"
#include "mqtt_client.h"
#include "nvs.h"
#include "nvs_flash.h"
#include "sdkconfig.h"
#include "soak_policy.h"
static const char*TAG="R06";
static atomic_bool ip,mqtt_online;
static atomic_uint_fast32_t link_dropped;
typedef struct{bool online;int64_t at_ms;} LinkNotice;
static QueueHandle_t link_queue;
static esp_mqtt_client_handle_t mqtt;
static void event(void*a,esp_event_base_t base,int32_t id,void*d){
 (void)a;(void)d;if(base==IP_EVENT&&id==IP_EVENT_STA_GOT_IP){atomic_store(&ip,true);LinkNotice n={true,esp_timer_get_time()/1000};if(xQueueSend(link_queue,&n,0)!=pdTRUE)atomic_fetch_add(&link_dropped,1);}
 if(base==WIFI_EVENT&&id==WIFI_EVENT_STA_DISCONNECTED){atomic_store(&ip,false);LinkNotice n={false,esp_timer_get_time()/1000};if(xQueueSend(link_queue,&n,0)!=pdTRUE)atomic_fetch_add(&link_dropped,1);}
}
static void mqtt_event(void*a,esp_event_base_t base,int32_t id,void*d){(void)a;(void)base;(void)d;if(id==MQTT_EVENT_CONNECTED)atomic_store(&mqtt_online,true);else if(id==MQTT_EVENT_DISCONNECTED)atomic_store(&mqtt_online,false);}
static bool start_network(void){
 if(strstr(CONFIG_R06_SSID,"REPLACE_")||strstr(CONFIG_R06_PASSWORD,"REPLACE_")||strstr(CONFIG_R06_BROKER,"REPLACE_")){ESP_LOGW(TAG,"CONFIG_MISSING: serial-only fixture; no Wi-Fi started");return false;}
 if(strlen(CONFIG_R06_SSID)>32||strlen(CONFIG_R06_PASSWORD)<8||strlen(CONFIG_R06_PASSWORD)>63||strncmp(CONFIG_R06_BROKER,"mqtt://",7)){ESP_LOGE(TAG,"CONFIG_INVALID: WPA2 SSID/password/trusted LAN MQTT expected");return false;}
 link_queue=xQueueCreate(32,sizeof(LinkNotice));if(!link_queue){ESP_LOGE(TAG,"LINK_QUEUE_ALLOC_FAILED");return false;}
 ESP_ERROR_CHECK(esp_netif_init());ESP_ERROR_CHECK(esp_event_loop_create_default());esp_netif_create_default_wifi_sta();wifi_init_config_t init=WIFI_INIT_CONFIG_DEFAULT();ESP_ERROR_CHECK(esp_wifi_init(&init));ESP_ERROR_CHECK(esp_wifi_set_storage(WIFI_STORAGE_RAM));
 ESP_ERROR_CHECK(esp_event_handler_register(WIFI_EVENT,ESP_EVENT_ANY_ID,event,NULL));ESP_ERROR_CHECK(esp_event_handler_register(IP_EVENT,IP_EVENT_STA_GOT_IP,event,NULL));wifi_config_t wi={0};memcpy(wi.sta.ssid,CONFIG_R06_SSID,strlen(CONFIG_R06_SSID));memcpy(wi.sta.password,CONFIG_R06_PASSWORD,strlen(CONFIG_R06_PASSWORD));wi.sta.threshold.authmode=WIFI_AUTH_WPA2_PSK;ESP_ERROR_CHECK(esp_wifi_set_mode(WIFI_MODE_STA));ESP_ERROR_CHECK(esp_wifi_set_config(WIFI_IF_STA,&wi));ESP_ERROR_CHECK(esp_wifi_start());
 esp_mqtt_client_config_t mc={.broker.address.uri=CONFIG_R06_BROKER,.network.reconnect_timeout_ms=2000,.outbox.limit=8192};mqtt=esp_mqtt_client_init(&mc);if(!mqtt){ESP_LOGE(TAG,"MQTT_ALLOC_FAILED");return true;}ESP_ERROR_CHECK(esp_mqtt_client_register_event(mqtt,ESP_EVENT_ANY_ID,mqtt_event,NULL));ESP_ERROR_CHECK(esp_mqtt_client_start(mqtt));return true;
}
static bool fault_enabled(void){
#if CONFIG_R06_FAULTS
 return true;
#else
 return false;
#endif
}
void app_main(void){
 esp_err_t e=nvs_flash_init();if(e!=ESP_OK){ESP_LOGE(TAG,"NVS=%s no auto erase",esp_err_to_name(e));return;}nvs_handle_t n;e=nvs_open("r06",NVS_READWRITE,&n);if(e!=ESP_OK){ESP_LOGE(TAG,"NVS_OPEN=%s",esp_err_to_name(e));return;}uint32_t boot=0;e=nvs_get_u32(n,"boots",&boot);if(e!=ESP_OK&&e!=ESP_ERR_NVS_NOT_FOUND){nvs_close(n);return;}if(boot==UINT32_MAX){ESP_LOGE(TAG,"BOOT_COUNTER_EXHAUSTED; preserve logs before deliberate reset");nvs_close(n);return;}boot++;e=nvs_set_u32(n,"boots",boot);if(e==ESP_OK)e=nvs_commit(n);nvs_close(n);if(e!=ESP_OK){ESP_LOGE(TAG,"BOOT_COMMIT=%s no boot success claimed",esp_err_to_name(e));return;}
 uint8_t mac[6];ESP_ERROR_CHECK(esp_read_mac(mac,ESP_MAC_WIFI_STA));char device[13];snprintf(device,sizeof(device),"%02x%02x%02x%02x%02x%02x",mac[0],mac[1],mac[2],mac[3],mac[4],mac[5]);char session[48];snprintf(session,sizeof(session),"%s-%"PRIu32"-%08"PRIx32,device,boot,esp_random());
 printf("BOOT,%s,%"PRIu32",%d\n",session,boot,esp_reset_reason());int64_t network_started=esp_timer_get_time()/1000;Recovery recovery={0};bool net=start_network();if(net)recovery_lost(&recovery,network_started);int64_t retry_at=0,manual_hold_until=0,pause_until=0,next_sample=esp_timer_get_time()/1000,next_status=next_sample;SampleRing ring={0};uint64_t seq=0,missed_ticks=0,submitted=0,rejected=0;bool corrupt=false;void*leak=NULL;
 fcntl(fileno(stdin),F_SETFL,O_NONBLOCK);char line[32]={0};unsigned used=0;bool discard=false;
 for(;;){
  int ch;while((ch=getchar())!=EOF){if(ch=='\r'||ch=='\n'){if(used&&!discard){line[used]=0;if(!fault_enabled())ESP_LOGW(TAG,"FAULTS_DISABLED");else if(!strcmp(line,"pause"))pause_until=esp_timer_get_time()/1000+10000;else if(!strcmp(line,"drop")&&net){manual_hold_until=esp_timer_get_time()/1000+5000;esp_wifi_disconnect();retry_at=manual_hold_until;}else if(!strcmp(line,"corrupt"))corrupt=true;else if(!strcmp(line,"leak")&&!leak){leak=heap_caps_malloc(4096,MALLOC_CAP_INTERNAL|MALLOC_CAP_8BIT);ESP_LOGW(TAG,"BOUNDED_LEAK_INJECTION allocated=%d; at most one4096B block",leak!=NULL);}else if(!strcmp(line,"release")){free(leak);leak=NULL;}else if(!strcmp(line,"reboot")){ESP_LOGW(TAG,"EXPLICIT_REBOOT_INJECTION");fflush(stdout);vTaskDelay(pdMS_TO_TICKS(100));esp_restart();}else ESP_LOGW(TAG,"UNKNOWN_OR_INAPPLICABLE_COMMAND");}used=0;discard=false;}else if(ch<32||ch>126||used>=31)discard=true;else if(!discard)line[used++]=(char)ch;}clearerr(stdin);
  int64_t now=esp_timer_get_time()/1000;bool online=atomic_load(&ip);LinkNotice notice;while(link_queue&&xQueueReceive(link_queue,&notice,0)==pdTRUE){if(notice.online){if(recovery_back(&recovery,notice.at_ms))printf("RECOVERY,%s,%"PRId64"\n",session,recovery.last_ms);}else{recovery_lost(&recovery,notice.at_ms);retry_at=notice.at_ms+2000;}}
  if(net&&!online&&now>=retry_at&&now>=manual_hold_until){retry_at=now+5000;e=esp_wifi_connect();if(e!=ESP_OK&&e!=ESP_ERR_WIFI_CONN)ESP_LOGW(TAG,"CONNECT_REQUEST=%s",esp_err_to_name(e));}
  uint64_t due=sample_due(now,&next_sample);if(due){seq+=due;missed_ticks+=due-1;Sample s={.seq=seq,.at_ms=now,.value=(int)(seq%100)};sample_push(&ring,s);}
  if(now>=pause_until){Sample s;if(sample_pop(&ring,&s)){char payload[192];int nbytes=snprintf(payload,sizeof(payload),"{\"session\":\"%s\",\"boot\":%"PRIu32",\"seq\":%"PRIu64",\"mono_ms\":%"PRId64",\"value\":%d}",session,boot,s.seq,s.at_ms,s.value);if(corrupt){snprintf(payload,sizeof(payload),"{synthetic-corrupt");nbytes=(int)strlen(payload);corrupt=false;}int mid=-1;if(mqtt&&atomic_load(&mqtt_online))mid=esp_mqtt_client_enqueue(mqtt,"hixinwu/r06/sample",payload,nbytes,1,0,true);if(mid>=0)submitted++;else rejected++;printf("SAMPLE,%s,%"PRIu32",%"PRIu64",%"PRId64",%d,%d\n",session,boot,s.seq,s.at_ms,s.value,mid);}}
  if(now>=next_status){next_status=now+5000;unsigned lost_events=(unsigned)atomic_load(&link_dropped);if(lost_events)printf("METRICS_INCOMPLETE,link_notice_dropped,%u\n",lost_events);printf("STATUS,%s,%"PRId64",%"PRIu64",%"PRIu64",%"PRIu64",%"PRIu64",%"PRIu64",%u,%u,%u,%u,%d,%d\n",session,now,seq,missed_ticks,ring.drop_new,submitted,rejected,(unsigned)esp_get_free_heap_size(),(unsigned)esp_get_minimum_free_heap_size(),(unsigned)heap_caps_get_largest_free_block(MALLOC_CAP_INTERNAL|MALLOC_CAP_8BIT),(unsigned)uxTaskGetStackHighWaterMark(NULL),online,atomic_load(&mqtt_online));}
  vTaskDelay(pdMS_TO_TICKS(20));
 }
}

// ---- main/soak_policy.h ----
/* SPDX-License-Identifier: MIT */
#pragma once
#include <stdbool.h>
#include <stdint.h>
#include <stddef.h>
#include <string.h>
typedef struct{uint64_t seq;int64_t at_ms;int value;} Sample;
typedef struct{Sample q[8];unsigned head,count;uint64_t drop_new;} SampleRing;
static inline bool sample_push(SampleRing*r,Sample s){if(r->count==8){r->drop_new++;return false;}r->q[(r->head+r->count)%8]=s;r->count++;return true;}
static inline bool sample_pop(SampleRing*r,Sample*out){if(!r->count)return false;*out=r->q[r->head];r->head=(r->head+1)%8;r->count--;return true;}
typedef struct{bool active;int64_t since,last_ms;uint64_t recoveries,max_ms;} Recovery;
static inline void recovery_lost(Recovery*r,int64_t now){if(!r->active){r->active=true;r->since=now;}}
static inline bool recovery_back(Recovery*r,int64_t now){if(!r->active)return false;if(now<r->since){r->active=false;return false;}r->last_ms=now-r->since;if((uint64_t)r->last_ms>r->max_ms)r->max_ms=(uint64_t)r->last_ms;r->recoveries++;r->active=false;return true;}
/* Keep monotonic64 across days; interval statistics never subtract wall clocks. */
static inline uint64_t sample_due(int64_t now,int64_t*next){if(now<*next)return 0;uint64_t n=(uint64_t)(now-*next)/1000+1;*next+=(int64_t)n*1000;return n;}

# ---- tools/receiver.py ----
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""Local observation receiver; commit valid/deduplicated records before callback return."""
import argparse,json,pathlib,re,sqlite3,time
SESSION=re.compile(r'[0-9a-f]{12}-[1-9][0-9]{0,9}-[0-9a-f]{8}\Z')
def decode(payload):
 if len(payload)>192:raise ValueError('length')
 d=json.loads(payload)
 if not isinstance(d,dict)or set(d)!={'session','boot','seq','mono_ms','value'}:raise ValueError('keys')
 if not isinstance(d['session'],str)or not SESSION.fullmatch(d['session']):raise ValueError('session')
 for key,lo,hi in [('boot',1,4294967295),('seq',1,9223372036854775807),('mono_ms',0,9223372036854775807),('value',0,99)]:
  if type(d[key])is not int or not lo<=d[key]<=hi:raise ValueError('range')
 if int(d['session'].split('-')[1])!=d['boot']:raise ValueError('boot identity')
 return d
class Store:
 def __init__(self,path):
  self.db=sqlite3.connect(path);self.db.execute('PRAGMA journal_mode=WAL');self.db.execute('PRAGMA synchronous=FULL');self.db.execute('CREATE TABLE IF NOT EXISTS samples(session TEXT,seq INTEGER,boot INTEGER,mono_ms INTEGER,value INTEGER,received_wall REAL,received_mono REAL,PRIMARY KEY(session,seq))');self.db.execute('CREATE TABLE IF NOT EXISTS counters(name TEXT PRIMARY KEY,n INTEGER NOT NULL)');self.db.commit()
 def ingest(self,payload):
  try:d=decode(payload)
  except (ValueError,UnicodeDecodeError):self.db.execute("INSERT INTO counters VALUES('invalid',1) ON CONFLICT(name)DO UPDATE SET n=n+1");self.db.commit();return 'INVALID'
  previous=self.db.execute('SELECT boot,mono_ms,value FROM samples WHERE session=? AND seq=?',(d['session'],d['seq'])).fetchone()
  if previous is not None:
   kind='duplicate'if previous==(d['boot'],d['mono_ms'],d['value'])else'conflict'
   self.db.execute('INSERT INTO counters VALUES(?,1) ON CONFLICT(name)DO UPDATE SET n=n+1',(kind,));self.db.commit();return kind.upper()
  self.db.execute('INSERT INTO samples VALUES(?,?,?,?,?,?,?)',(d['session'],d['seq'],d['boot'],d['mono_ms'],d['value'],time.time(),time.monotonic()));self.db.commit();return 'COMMITTED'
 def close(self):self.db.close()
def main():
 import paho.mqtt.client as mqtt
 p=argparse.ArgumentParser();p.add_argument('--host',default='127.0.0.1');p.add_argument('--port',type=int,default=1883);p.add_argument('--db',required=True);a=p.parse_args();store=Store(a.db);client=mqtt.Client(client_id='hixinwu-r06-observer',clean_session=True)
 def connected(c,u,f,rc):
  if rc!=0:raise RuntimeError('broker connect rejected')
  result,mid=c.subscribe('hixinwu/r06/sample',1)
  if result!=mqtt.MQTT_ERR_SUCCESS:raise RuntimeError('subscribe rejected')
  print('OBSERVER_CONNECTED: subscription requested; host observation only',flush=True)
 def message(c,u,msg):
  if msg.retain:
   store.db.execute("INSERT INTO counters VALUES('retained',1) ON CONFLICT(name)DO UPDATE SET n=n+1");store.db.commit();print('HOST_RETAINED_REJECTED',flush=True);return
  try:result=store.ingest(msg.payload)
  except sqlite3.Error:c.disconnect();raise
  print('HOST_'+result,flush=True) # payload/token values deliberately not logged
 client.on_connect=connected;client.on_message=message;client.on_subscribe=lambda c,u,mid,q:print('OBSERVER_READY: broker SUBACK received',flush=True);client.connect(a.host,a.port,30)
 try:client.loop_forever()
 except KeyboardInterrupt:pass
 finally:client.disconnect();store.close()
if __name__=='__main__':main()

# ---- tools/analyze.py ----
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""Observed serial window versus committed receiver IDs; no soak pass stamp."""
import argparse,pathlib,csv,json,sqlite3,re
SESSION=re.compile(r'[0-9a-f]{12}-[1-9][0-9]{0,9}-[0-9a-f]{8}\Z')
I64=9223372036854775807;U64=18446744073709551615;U32=4294967295
def number(text,lo=0,hi=I64):
 if not re.fullmatch(r'-?(?:0|[1-9][0-9]*)',text):raise ValueError('integer encoding')
 value=int(text)
 if not lo<=value<=hi:raise ValueError('integer range')
 return value
def analyze(serial,database):
 sessions={};recoveries=[];warnings=[]
 for line in pathlib.Path(serial).read_text().splitlines():
  row=next(csv.reader([line]))
  if not row or row[0]not in('BOOT','SAMPLE','STATUS','RECOVERY','METRICS_INCOMPLETE'):continue
  if row[0]=='METRICS_INCOMPLETE':
   if len(row)!=3 or row[1]!='link_notice_dropped':raise ValueError('metrics warning fields')
   warnings.append({'kind':row[1],'dropped':number(row[2],1,U32)});continue
  if len(row)<2 or not SESSION.fullmatch(row[1]):raise ValueError('session')
  name=row[1];boot=number(name.split('-')[1],1,U32)
  state=sessions.setdefault(name,{'bootSeen':False,'maxSeqObserved':0,'serialSampleIDs':set(),'maxMonoMs':0,'heapSamples':[]})
  if row[0]=='BOOT'and len(row)==4:
   if number(row[2],1,U32)!=boot:raise ValueError('boot identity')
   number(row[3],0,U32);state['bootSeen']=True;state['bootCounter']=boot
  elif row[0]=='SAMPLE'and len(row)==7:
   if number(row[2],1,U32)!=boot:raise ValueError('sample boot identity')
   seq=number(row[3],1);mono=number(row[4]);number(row[5],0,99);number(row[6],-2,65535)
   state['serialSampleIDs'].add(seq);state['maxSeqObserved']=max(state['maxSeqObserved'],seq);state['maxMonoMs']=max(state['maxMonoMs'],mono)
  elif row[0]=='STATUS'and len(row)==14:
   mono=number(row[2]);seq=number(row[3]);[number(x,0,U64)for x in row[4:8]];[number(x,0,U32)for x in row[8:12]];number(row[12],0,1);number(row[13],0,1)
   state['maxSeqObserved']=max(state['maxSeqObserved'],seq);state['maxMonoMs']=max(state['maxMonoMs'],mono);state['heapSamples'].append({'mono_ms':mono,'free':int(row[8]),'min':int(row[9]),'largest':int(row[10])})
  elif row[0]=='RECOVERY'and len(row)==3:recoveries.append({'session':name,'ms':number(row[2])})
  else:raise ValueError('field count')
 db=sqlite3.connect('file:'+str(pathlib.Path(database).resolve())+'?mode=ro',uri=True);report=[]
 try:
  for name,state in sessions.items():
   ids={r[0]for r in db.execute('SELECT seq FROM samples WHERE session=?',(name,))};den=state['maxSeqObserved'];delivered=len({i for i in ids if 1<=i<=den});serialids=state.pop('serialSampleIDs')
   state.update({'session':name,'serialSamplesObserved':len(serialids),'receiverCommittedWithinObservedWindow':delivered,'expectedSequenceWindow':den,'unobservedAtReceiver':den-delivered if state['bootSeen']else None,'unobservedFraction':(den-delivered)/den if den and state['bootSeen']else None});report.append(state)
  return{'scope':'Observed windows only. Complete capture from BOOT plus trusted receiver needed; serial loss/reboots outside log and tail after last STATUS are not proved. No 24h/7d success, power, Wi-Fi or real sensor result inferred. Compare same workload and fault schedule, not just minimum heap.','sessions':report,'wifiRecoveryNotMQTTOrDelivery':recoveries,'linkRecoveryMetricsIncomplete':bool(warnings),'metricsWarnings':warnings,'receiverCounters':dict(db.execute('SELECT name,n FROM counters'))}
 finally:db.close()
def main():
 p=argparse.ArgumentParser();p.add_argument('--serial',required=True);p.add_argument('--db',required=True);a=p.parse_args()
 try:result=analyze(a.serial,a.db)
 except(ValueError,csv.Error,sqlite3.Error)as error:p.error('malformed observation/database: '+str(error)+'; original logs retained, no silent acceptance')
 print(json.dumps(result,indent=2))
if __name__=='__main__':main()

# POSIX shell · 主机环境已安装tools/requirements.txt
cc -std=c11 -Wall -Wextra -Werror -fsanitize=address,undefined tests/host.c -o /tmp/r06-host
/tmp/r06-host
python tests/event_body.py --out /tmp/r06-event-proof
python tests/tools_fixture.py --out /tmp/r06-local-proof --broker /ABS/PATH/TO/mosquitto